[dpdk-dev] [PATCH] net/af_xdp: free mbuf when allocate Tx queue fails

[dpdk-dev] [PATCH] net/af_xdp: free mbuf when allocate Tx queue fails

[Xiaolong Ye]

When we fail to allocate enough slots in tx queue for transmitting
packets, we need to free the corresponding mbufs.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..bc7973b56 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,7 +276,8 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
-		return 0;
+		nb_pkts = 0;
+		goto out;
 	}
 
 	for (i = 0; i < nb_pkts; i++) {
@@ -296,7 +297,6 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 			valid++;
 			tx_bytes += mbuf->pkt_len;
 		}
-		rte_pktmbuf_free(mbuf);
 	}
 
 	xsk_ring_prod__submit(&txq->tx, nb_pkts);
@@ -311,6 +311,10 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	txq->stats.tx_pkts += valid;
 	txq->stats.tx_bytes += tx_bytes;
 
+ out:
+	for (i = 0; i < nb_pkts; i++)
+		rte_pktmbuf_free(bufs[i]);
+
 	return nb_pkts;
 }
 
-- 
2.17.1

[dpdk-dev] [PATCH] net/af_xdp: free mbuf when allocate Tx queue fails

[Xiaolong Ye]

When we fail to allocate enough slots in tx queue for transmitting
packets, we need to free the corresponding mbufs.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..bc7973b56 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,7 +276,8 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
-		return 0;
+		nb_pkts = 0;
+		goto out;
 	}
 
 	for (i = 0; i < nb_pkts; i++) {
@@ -296,7 +297,6 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 			valid++;
 			tx_bytes += mbuf->pkt_len;
 		}
-		rte_pktmbuf_free(mbuf);
 	}
 
 	xsk_ring_prod__submit(&txq->tx, nb_pkts);
@@ -311,6 +311,10 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	txq->stats.tx_pkts += valid;
 	txq->stats.tx_bytes += tx_bytes;
 
+ out:
+	for (i = 0; i < nb_pkts; i++)
+		rte_pktmbuf_free(bufs[i]);
+
 	return nb_pkts;
 }
 
-- 
2.17.1

Re: [dpdk-dev] [PATCH] net/af_xdp: free mbuf when allocate Tx queue fails

[David Marchand]

On Tue, Apr 9, 2019 at 10:27 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When we fail to allocate enough slots in tx queue for transmitting
> packets, we need to free the corresponding mbufs.
>

You'd better let the application retry on its own...


> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 8 ++++++--
>  1 file changed, 6 insertions(+), 2 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..bc7973b56 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,7 +276,8 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> -               return 0;
> +               nb_pkts = 0;
> +               goto out;
>

...and free back the buffers you got from the umem ring here.

        }
>
>         for (i = 0; i < nb_pkts; i++) {
>


-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: free mbuf when allocate Tx queue fails

[David Marchand]

On Tue, Apr 9, 2019 at 10:27 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When we fail to allocate enough slots in tx queue for transmitting
> packets, we need to free the corresponding mbufs.
>

You'd better let the application retry on its own...


> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 8 ++++++--
>  1 file changed, 6 insertions(+), 2 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..bc7973b56 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,7 +276,8 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> -               return 0;
> +               nb_pkts = 0;
> +               goto out;
>

...and free back the buffers you got from the umem ring here.

        }
>
>         for (i = 0; i < nb_pkts; i++) {
>


-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: free mbuf when allocate Tx queue fails

[Ye Xiaolong]

Hi, David

On 04/09, David Marchand wrote:
>On Tue, Apr 9, 2019 at 10:27 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> When we fail to allocate enough slots in tx queue for transmitting
>> packets, we need to free the corresponding mbufs.
>>
>
>You'd better let the application retry on its own...

Make sense.

>
>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 8 ++++++--
>>  1 file changed, 6 insertions(+), 2 deletions(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 007a1c6b4..bc7973b56 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -276,7 +276,8 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
>> {
>>                 kick_tx(txq);
>> -               return 0;
>> +               nb_pkts = 0;
>> +               goto out;
>>
>
>...and free back the buffers you got from the umem ring here.

Agree, will send a new version.

Thanks,
Xiaolong
>
>        }
>>
>>         for (i = 0; i < nb_pkts; i++) {
>>
>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: free mbuf when allocate Tx queue fails

[Ye Xiaolong]

Hi, David

On 04/09, David Marchand wrote:
>On Tue, Apr 9, 2019 at 10:27 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> When we fail to allocate enough slots in tx queue for transmitting
>> packets, we need to free the corresponding mbufs.
>>
>
>You'd better let the application retry on its own...

Make sense.

>
>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 8 ++++++--
>>  1 file changed, 6 insertions(+), 2 deletions(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 007a1c6b4..bc7973b56 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -276,7 +276,8 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
>> {
>>                 kick_tx(txq);
>> -               return 0;
>> +               nb_pkts = 0;
>> +               goto out;
>>
>
>...and free back the buffers you got from the umem ring here.

Agree, will send a new version.

Thanks,
Xiaolong
>
>        }
>>
>>         for (i = 0; i < nb_pkts; i++) {
>>
>
>
>-- 
>David Marchand

[dpdk-dev] [PATCH] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

Re: [dpdk-dev] [PATCH] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[David Marchand]

On Tue, Apr 9, 2019 at 5:24 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When it fails to allocate enough slots in Tx queue for transmitting
> packets, we need to return the dequeued addrs to buf ring.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>  1 file changed, 1 insertion(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..5cc643ce2 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
>                 return 0;
>         }
>
> --
> 2.17.1
>

Looks good to me.
But I have an additional question.

After the for loop that only picks the mbufs it can copy to the xdp desc,
is it normal to call xsk_ring_prod__submit(&txq->tx, nb_pkts); rather than
with valid count ?

-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[David Marchand]

On Tue, Apr 9, 2019 at 5:24 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When it fails to allocate enough slots in Tx queue for transmitting
> packets, we need to return the dequeued addrs to buf ring.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>  1 file changed, 1 insertion(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..5cc643ce2 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
>                 return 0;
>         }
>
> --
> 2.17.1
>

Looks good to me.
But I have an additional question.

After the for loop that only picks the mbufs it can copy to the xdp desc,
is it normal to call xsk_ring_prod__submit(&txq->tx, nb_pkts); rather than
with valid count ?

-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Ye Xiaolong]

On 04/10, David Marchand wrote:
>On Tue, Apr 9, 2019 at 5:24 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> When it fails to allocate enough slots in Tx queue for transmitting
>> packets, we need to return the dequeued addrs to buf ring.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>>  1 file changed, 1 insertion(+)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 007a1c6b4..5cc643ce2 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
>> {
>>                 kick_tx(txq);
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
>> NULL);
>>                 return 0;
>>         }
>>
>> --
>> 2.17.1
>>
>
>Looks good to me.
>But I have an additional question.
>
>After the for loop that only picks the mbufs it can copy to the xdp desc,
>is it normal to call xsk_ring_prod__submit(&txq->tx, nb_pkts); rather than
>with valid count ?

Good catch, I think it needs to submit valid count other than nb_pkts, and return back
(nb_pkts - count) addrs to buf ring.

I'll send a fix patch for it.

Thanks,
Xiaolong
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Ye Xiaolong]

On 04/10, David Marchand wrote:
>On Tue, Apr 9, 2019 at 5:24 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> When it fails to allocate enough slots in Tx queue for transmitting
>> packets, we need to return the dequeued addrs to buf ring.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>>  1 file changed, 1 insertion(+)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 007a1c6b4..5cc643ce2 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
>> {
>>                 kick_tx(txq);
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
>> NULL);
>>                 return 0;
>>         }
>>
>> --
>> 2.17.1
>>
>
>Looks good to me.
>But I have an additional question.
>
>After the for loop that only picks the mbufs it can copy to the xdp desc,
>is it normal to call xsk_ring_prod__submit(&txq->tx, nb_pkts); rather than
>with valid count ?

Good catch, I think it needs to submit valid count other than nb_pkts, and return back
(nb_pkts - count) addrs to buf ring.

I'll send a fix patch for it.

Thanks,
Xiaolong
>
>-- 
>David Marchand

[dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[Xiaolong Ye]

Since tx callback only picks mbufs that can be copied to the xdp desc, it
should call xsk_ring_prod__submit with valid count other than nb_pkts.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reported-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..8c2ba5740 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -300,7 +300,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_free(mbuf);
 	}
 
-	xsk_ring_prod__submit(&txq->tx, nb_pkts);
+	xsk_ring_prod__submit(&txq->tx, valid);
 
 	kick_tx(txq);
 
-- 
2.17.1

[dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[Xiaolong Ye]

Since tx callback only picks mbufs that can be copied to the xdp desc, it
should call xsk_ring_prod__submit with valid count other than nb_pkts.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reported-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..8c2ba5740 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -300,7 +300,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_free(mbuf);
 	}
 
-	xsk_ring_prod__submit(&txq->tx, nb_pkts);
+	xsk_ring_prod__submit(&txq->tx, valid);
 
 	kick_tx(txq);
 
-- 
2.17.1

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[David Marchand]

On Wed, Apr 10, 2019 at 12:58 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> Since tx callback only picks mbufs that can be copied to the xdp desc, it
> should call xsk_ring_prod__submit with valid count other than nb_pkts.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Reported-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 2 +-
>  1 file changed, 1 insertion(+), 1 deletion(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..8c2ba5740 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -300,7 +300,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>                 rte_pktmbuf_free(mbuf);
>         }
>
> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
> +       xsk_ring_prod__submit(&txq->tx, valid);
>

Err, well, I think we will have an issue here.

Taking from the 5.1.0-rc4 sources I have:

static inline __u32 xsk_prod_nb_free(struct xsk_ring_prod *r, __u32 nb)
{
        __u32 free_entries = r->cached_cons - r->cached_prod;

        if (free_entries >= nb)
                return free_entries;

        /* Refresh the local tail pointer.
         * cached_cons is r->size bigger than the real consumer pointer so
         * that this addition can be avoided in the more frequently
         * executed code that computs free_entries in the beginning of
         * this function. Without this optimization it whould have been
         * free_entries = r->cached_prod - r->cached_cons + r->size.
         */
        r->cached_cons = *r->consumer + r->size;

        return r->cached_cons - r->cached_prod;
}

static inline size_t xsk_ring_prod__reserve(struct xsk_ring_prod *prod,
                                            size_t nb, __u32 *idx)
{
        if (unlikely(xsk_prod_nb_free(prod, nb) < nb))
                return 0;

        *idx = prod->cached_prod;
        prod->cached_prod += nb;

        return nb;
}

static inline void xsk_ring_prod__submit(struct xsk_ring_prod *prod, size_t
nb)
{
        /* Make sure everything has been written to the ring before
signalling
         * this to the kernel.
         */
        smp_wmb();

        *prod->producer += nb;
}


If we reserve N slots, but only submit n slots, we end up with an incorrect
opinion of the number of available slots later.
Either xsk_ring_prod__submit should also update cached_prod (but I am not
sure it was the intent of this api), or we must ensure that both reserve
and submit are consistent.

Did I miss some subtility ?


-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[David Marchand]

On Wed, Apr 10, 2019 at 12:58 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> Since tx callback only picks mbufs that can be copied to the xdp desc, it
> should call xsk_ring_prod__submit with valid count other than nb_pkts.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Reported-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 2 +-
>  1 file changed, 1 insertion(+), 1 deletion(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..8c2ba5740 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -300,7 +300,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>                 rte_pktmbuf_free(mbuf);
>         }
>
> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
> +       xsk_ring_prod__submit(&txq->tx, valid);
>

Err, well, I think we will have an issue here.

Taking from the 5.1.0-rc4 sources I have:

static inline __u32 xsk_prod_nb_free(struct xsk_ring_prod *r, __u32 nb)
{
        __u32 free_entries = r->cached_cons - r->cached_prod;

        if (free_entries >= nb)
                return free_entries;

        /* Refresh the local tail pointer.
         * cached_cons is r->size bigger than the real consumer pointer so
         * that this addition can be avoided in the more frequently
         * executed code that computs free_entries in the beginning of
         * this function. Without this optimization it whould have been
         * free_entries = r->cached_prod - r->cached_cons + r->size.
         */
        r->cached_cons = *r->consumer + r->size;

        return r->cached_cons - r->cached_prod;
}

static inline size_t xsk_ring_prod__reserve(struct xsk_ring_prod *prod,
                                            size_t nb, __u32 *idx)
{
        if (unlikely(xsk_prod_nb_free(prod, nb) < nb))
                return 0;

        *idx = prod->cached_prod;
        prod->cached_prod += nb;

        return nb;
}

static inline void xsk_ring_prod__submit(struct xsk_ring_prod *prod, size_t
nb)
{
        /* Make sure everything has been written to the ring before
signalling
         * this to the kernel.
         */
        smp_wmb();

        *prod->producer += nb;
}


If we reserve N slots, but only submit n slots, we end up with an incorrect
opinion of the number of available slots later.
Either xsk_ring_prod__submit should also update cached_prod (but I am not
sure it was the intent of this api), or we must ensure that both reserve
and submit are consistent.

Did I miss some subtility ?


-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[Ye Xiaolong]

On 04/10, David Marchand wrote:
>On Wed, Apr 10, 2019 at 12:58 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> Since tx callback only picks mbufs that can be copied to the xdp desc, it
>> should call xsk_ring_prod__submit with valid count other than nb_pkts.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Reported-by: David Marchand <david.marchand@redhat.com>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 2 +-
>>  1 file changed, 1 insertion(+), 1 deletion(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..8c2ba5740 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -300,7 +300,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>                 rte_pktmbuf_free(mbuf);
>>         }
>>
>> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
>> +       xsk_ring_prod__submit(&txq->tx, valid);
>>
>
>Err, well, I think we will have an issue here.
>
>Taking from the 5.1.0-rc4 sources I have:
>
>static inline __u32 xsk_prod_nb_free(struct xsk_ring_prod *r, __u32 nb)
>{
>        __u32 free_entries = r->cached_cons - r->cached_prod;
>
>        if (free_entries >= nb)
>                return free_entries;
>
>        /* Refresh the local tail pointer.
>         * cached_cons is r->size bigger than the real consumer pointer so
>         * that this addition can be avoided in the more frequently
>         * executed code that computs free_entries in the beginning of
>         * this function. Without this optimization it whould have been
>         * free_entries = r->cached_prod - r->cached_cons + r->size.
>         */
>        r->cached_cons = *r->consumer + r->size;
>
>        return r->cached_cons - r->cached_prod;
>}
>
>static inline size_t xsk_ring_prod__reserve(struct xsk_ring_prod *prod,
>                                            size_t nb, __u32 *idx)
>{
>        if (unlikely(xsk_prod_nb_free(prod, nb) < nb))
>                return 0;
>
>        *idx = prod->cached_prod;
>        prod->cached_prod += nb;
>
>        return nb;
>}
>
>static inline void xsk_ring_prod__submit(struct xsk_ring_prod *prod, size_t
>nb)
>{
>        /* Make sure everything has been written to the ring before
>signalling
>         * this to the kernel.
>         */
>        smp_wmb();
>
>        *prod->producer += nb;
>}
>
>
>If we reserve N slots, but only submit n slots, we end up with an incorrect
>opinion of the number of available slots later.
>Either xsk_ring_prod__submit should also update cached_prod (but I am not
>sure it was the intent of this api), or we must ensure that both reserve
>and submit are consistent.

I think you are right, current design does have the flaw, I haven't thought of
it before :( So in order to make sure both reserve and submit are consistent, what
about we check the valid count of mbuf at the beginning, then reserve the valid 
count slots?

Thanks,
Xiaolong
>
>Did I miss some subtility ?
>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[Ye Xiaolong]

On 04/10, David Marchand wrote:
>On Wed, Apr 10, 2019 at 12:58 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> Since tx callback only picks mbufs that can be copied to the xdp desc, it
>> should call xsk_ring_prod__submit with valid count other than nb_pkts.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Reported-by: David Marchand <david.marchand@redhat.com>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 2 +-
>>  1 file changed, 1 insertion(+), 1 deletion(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..8c2ba5740 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -300,7 +300,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>                 rte_pktmbuf_free(mbuf);
>>         }
>>
>> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
>> +       xsk_ring_prod__submit(&txq->tx, valid);
>>
>
>Err, well, I think we will have an issue here.
>
>Taking from the 5.1.0-rc4 sources I have:
>
>static inline __u32 xsk_prod_nb_free(struct xsk_ring_prod *r, __u32 nb)
>{
>        __u32 free_entries = r->cached_cons - r->cached_prod;
>
>        if (free_entries >= nb)
>                return free_entries;
>
>        /* Refresh the local tail pointer.
>         * cached_cons is r->size bigger than the real consumer pointer so
>         * that this addition can be avoided in the more frequently
>         * executed code that computs free_entries in the beginning of
>         * this function. Without this optimization it whould have been
>         * free_entries = r->cached_prod - r->cached_cons + r->size.
>         */
>        r->cached_cons = *r->consumer + r->size;
>
>        return r->cached_cons - r->cached_prod;
>}
>
>static inline size_t xsk_ring_prod__reserve(struct xsk_ring_prod *prod,
>                                            size_t nb, __u32 *idx)
>{
>        if (unlikely(xsk_prod_nb_free(prod, nb) < nb))
>                return 0;
>
>        *idx = prod->cached_prod;
>        prod->cached_prod += nb;
>
>        return nb;
>}
>
>static inline void xsk_ring_prod__submit(struct xsk_ring_prod *prod, size_t
>nb)
>{
>        /* Make sure everything has been written to the ring before
>signalling
>         * this to the kernel.
>         */
>        smp_wmb();
>
>        *prod->producer += nb;
>}
>
>
>If we reserve N slots, but only submit n slots, we end up with an incorrect
>opinion of the number of available slots later.
>Either xsk_ring_prod__submit should also update cached_prod (but I am not
>sure it was the intent of this api), or we must ensure that both reserve
>and submit are consistent.

I think you are right, current design does have the flaw, I haven't thought of
it before :( So in order to make sure both reserve and submit are consistent, what
about we check the valid count of mbuf at the beginning, then reserve the valid 
count slots?

Thanks,
Xiaolong
>
>Did I miss some subtility ?
>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[David Marchand]

Hello Xiaolong,

On Thu, Apr 11, 2019 at 4:30 AM Ye Xiaolong <xiaolong.ye@intel.com> wrote:

> On 04/10, David Marchand wrote:
> >If we reserve N slots, but only submit n slots, we end up with an
> incorrect
> >opinion of the number of available slots later.
> >Either xsk_ring_prod__submit should also update cached_prod (but I am not
> >sure it was the intent of this api), or we must ensure that both reserve
> >and submit are consistent.
>
> I think you are right, current design does have the flaw, I haven't
> thought of
> it before :( So in order to make sure both reserve and submit are
> consistent, what
> about we check the valid count of mbuf at the beginning, then reserve the
> valid
> count slots?
>
>
Ok, I can see other places to inspect in the driver: reserve_fill_queue()
for the same issue, and eth_af_xdp_rx() for a similar issue but with
xsk_ring_cons__peek()/xsk_ring_cons__release() ?
Validating the needed slots count before reserving/peeking in the prod/cons
rings seems the most simple fix.


-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[David Marchand]

Hello Xiaolong,

On Thu, Apr 11, 2019 at 4:30 AM Ye Xiaolong <xiaolong.ye@intel.com> wrote:

> On 04/10, David Marchand wrote:
> >If we reserve N slots, but only submit n slots, we end up with an
> incorrect
> >opinion of the number of available slots later.
> >Either xsk_ring_prod__submit should also update cached_prod (but I am not
> >sure it was the intent of this api), or we must ensure that both reserve
> >and submit are consistent.
>
> I think you are right, current design does have the flaw, I haven't
> thought of
> it before :( So in order to make sure both reserve and submit are
> consistent, what
> about we check the valid count of mbuf at the beginning, then reserve the
> valid
> count slots?
>
>
Ok, I can see other places to inspect in the driver: reserve_fill_queue()
for the same issue, and eth_af_xdp_rx() for a similar issue but with
xsk_ring_cons__peek()/xsk_ring_cons__release() ?
Validating the needed slots count before reserving/peeking in the prod/cons
rings seems the most simple fix.


-- 
David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[Ye Xiaolong]

On 04/11, David Marchand wrote:
>Hello Xiaolong,
>
>On Thu, Apr 11, 2019 at 4:30 AM Ye Xiaolong <xiaolong.ye@intel.com> wrote:
>
>> On 04/10, David Marchand wrote:
>> >If we reserve N slots, but only submit n slots, we end up with an
>> incorrect
>> >opinion of the number of available slots later.
>> >Either xsk_ring_prod__submit should also update cached_prod (but I am not
>> >sure it was the intent of this api), or we must ensure that both reserve
>> >and submit are consistent.
>>
>> I think you are right, current design does have the flaw, I haven't
>> thought of
>> it before :( So in order to make sure both reserve and submit are
>> consistent, what
>> about we check the valid count of mbuf at the beginning, then reserve the
>> valid
>> count slots?
>>
>>
>Ok, I can see other places to inspect in the driver: reserve_fill_queue()
>for the same issue, and eth_af_xdp_rx() for a similar issue but with
>xsk_ring_cons__peek()/xsk_ring_cons__release() ?
>Validating the needed slots count before reserving/peeking in the prod/cons
>rings seems the most simple fix.

Ok, then I'll handle them in a following patch.

Thanks,
Xiaolong
>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH] net/af_xdp: submit valid count to Tx queue

[Ye Xiaolong]

On 04/11, David Marchand wrote:
>Hello Xiaolong,
>
>On Thu, Apr 11, 2019 at 4:30 AM Ye Xiaolong <xiaolong.ye@intel.com> wrote:
>
>> On 04/10, David Marchand wrote:
>> >If we reserve N slots, but only submit n slots, we end up with an
>> incorrect
>> >opinion of the number of available slots later.
>> >Either xsk_ring_prod__submit should also update cached_prod (but I am not
>> >sure it was the intent of this api), or we must ensure that both reserve
>> >and submit are consistent.
>>
>> I think you are right, current design does have the flaw, I haven't
>> thought of
>> it before :( So in order to make sure both reserve and submit are
>> consistent, what
>> about we check the valid count of mbuf at the beginning, then reserve the
>> valid
>> count slots?
>>
>>
>Ok, I can see other places to inspect in the driver: reserve_fill_queue()
>for the same issue, and eth_af_xdp_rx() for a similar issue but with
>xsk_ring_cons__peek()/xsk_ring_cons__release() ?
>Validating the needed slots count before reserving/peeking in the prod/cons
>rings seems the most simple fix.

Ok, then I'll handle them in a following patch.

Thanks,
Xiaolong
>
>
>-- 
>David Marchand

[dpdk-dev] [PATCH 1/2] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH 1/2] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots, but only submit n slots,
we would end up with an incorrect opinion of the number of available slots
later, we also would get wrong idx when we call xsk_ring_prod__reserve next
time. It also applies to xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reported-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 80 +++++++++++++++--------------
 1 file changed, 41 insertions(+), 39 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..76a6a8331 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -138,22 +138,19 @@ reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
 	uint32_t idx;
-	int i, ret;
-
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
-		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
-	}
+	int i;
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
 		void *addr = NULL;
 		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
 			break;
 		}
-		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
+		if (unlikely(!xsk_ring_prod__reserve(fq, 1, &idx))) {
+			AF_XDP_LOG(WARNING, "Failed to reserve 1 fq desc.\n");
+			break;
+		}
+		fq_addr = xsk_ring_prod__fill_addr(fq, idx);
 		*fq_addr = (uint64_t)addr;
 	}
 
@@ -179,6 +176,9 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
 		return 0;
@@ -186,9 +186,6 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -211,6 +208,10 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	xsk_ring_cons__release(rx, rcvd);
 
+	/* free the extra mbufs */
+	for (; rcvd < nb_pkts; rcvd++)
+		rte_pktmbuf_free(mbufs[rcvd]);
+
 	/* statistics */
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
@@ -261,55 +262,56 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct xsk_umem_info *umem = txq->pair->umem;
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
+	struct rte_mbuf *valid_bufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
+	uint16_t nb_valid = 0;
 	uint32_t idx_tx;
+	uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
 	pull_umem_cq(umem, nb_pkts);
 
-	nb_pkts = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
-					nb_pkts, NULL);
-	if (nb_pkts == 0)
+	for (i = 0; i < nb_pkts; i++) {
+		if (bufs[i]->pkt_len <= buf_len)
+			valid_bufs[nb_valid++] = bufs[i];
+		else
+			rte_pktmbuf_free(bufs[i]);
+	}
+
+	nb_valid = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
+					nb_valid, NULL);
+	if (nb_valid == 0)
 		return 0;
 
-	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
+	if (xsk_ring_prod__reserve(&txq->tx, nb_valid, &idx_tx) != nb_valid) {
 		kick_tx(txq);
-		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_valid, NULL);
 		return 0;
 	}
 
-	for (i = 0; i < nb_pkts; i++) {
+	for (i = 0; i < nb_valid; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
-		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+		mbuf = valid_bufs[i];
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
 		rte_pktmbuf_free(mbuf);
 	}
 
-	xsk_ring_prod__submit(&txq->tx, nb_pkts);
+	xsk_ring_prod__submit(&txq->tx, nb_valid);
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.err_pkts += nb_pkts - nb_valid;
+	txq->stats.tx_pkts += nb_valid;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

[dpdk-dev] [PATCH 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots, but only submit n slots,
we would end up with an incorrect opinion of the number of available slots
later, we also would get wrong idx when we call xsk_ring_prod__reserve next
time. It also applies to xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reported-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 80 +++++++++++++++--------------
 1 file changed, 41 insertions(+), 39 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..76a6a8331 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -138,22 +138,19 @@ reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
 	uint32_t idx;
-	int i, ret;
-
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
-		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
-	}
+	int i;
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
 		void *addr = NULL;
 		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
 			break;
 		}
-		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
+		if (unlikely(!xsk_ring_prod__reserve(fq, 1, &idx))) {
+			AF_XDP_LOG(WARNING, "Failed to reserve 1 fq desc.\n");
+			break;
+		}
+		fq_addr = xsk_ring_prod__fill_addr(fq, idx);
 		*fq_addr = (uint64_t)addr;
 	}
 
@@ -179,6 +176,9 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
 		return 0;
@@ -186,9 +186,6 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -211,6 +208,10 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	xsk_ring_cons__release(rx, rcvd);
 
+	/* free the extra mbufs */
+	for (; rcvd < nb_pkts; rcvd++)
+		rte_pktmbuf_free(mbufs[rcvd]);
+
 	/* statistics */
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
@@ -261,55 +262,56 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct xsk_umem_info *umem = txq->pair->umem;
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
+	struct rte_mbuf *valid_bufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
+	uint16_t nb_valid = 0;
 	uint32_t idx_tx;
+	uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
 	pull_umem_cq(umem, nb_pkts);
 
-	nb_pkts = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
-					nb_pkts, NULL);
-	if (nb_pkts == 0)
+	for (i = 0; i < nb_pkts; i++) {
+		if (bufs[i]->pkt_len <= buf_len)
+			valid_bufs[nb_valid++] = bufs[i];
+		else
+			rte_pktmbuf_free(bufs[i]);
+	}
+
+	nb_valid = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
+					nb_valid, NULL);
+	if (nb_valid == 0)
 		return 0;
 
-	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
+	if (xsk_ring_prod__reserve(&txq->tx, nb_valid, &idx_tx) != nb_valid) {
 		kick_tx(txq);
-		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_valid, NULL);
 		return 0;
 	}
 
-	for (i = 0; i < nb_pkts; i++) {
+	for (i = 0; i < nb_valid; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
-		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+		mbuf = valid_bufs[i];
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
 		rte_pktmbuf_free(mbuf);
 	}
 
-	xsk_ring_prod__submit(&txq->tx, nb_pkts);
+	xsk_ring_prod__submit(&txq->tx, nb_valid);
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.err_pkts += nb_pkts - nb_valid;
+	txq->stats.tx_pkts += nb_valid;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

Re: [dpdk-dev] [PATCH 2/2] net/af_xdp: make reserve/submit peek/release consistent

[David Marchand]

On Fri, Apr 12, 2019 at 4:54 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> As David pointed out, if we reserve N slots, but only submit n slots,
> we would end up with an incorrect opinion of the number of available slots
> later, we also would get wrong idx when we call xsk_ring_prod__reserve next
> time. It also applies to xsk_ring_cons__peek()/xsk_ring_cons__release().
>
> This patch ensures that both reserve/submit and peek/release are
> consistent.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Reported-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 80 +++++++++++++++--------------
>  1 file changed, 41 insertions(+), 39 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..76a6a8331 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -138,22 +138,19 @@ reserve_fill_queue(struct xsk_umem_info *umem, int
> reserve_size)
>  {
>         struct xsk_ring_prod *fq = &umem->fq;
>         uint32_t idx;
> -       int i, ret;
> -
> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
> -       if (unlikely(!ret)) {
> -               AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
> -               return ret;
> -       }
> +       int i;
>
>         for (i = 0; i < reserve_size; i++) {
>                 __u64 *fq_addr;
>                 void *addr = NULL;
>                 if (rte_ring_dequeue(umem->buf_ring, &addr)) {
> -                       i--;
>                         break;
>                 }
> -               fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
> +               if (unlikely(!xsk_ring_prod__reserve(fq, 1, &idx))) {
> +                       AF_XDP_LOG(WARNING, "Failed to reserve 1 fq
> desc.\n");
> +                       break;
> +               }
> +               fq_addr = xsk_ring_prod__fill_addr(fq, idx);
>                 *fq_addr = (uint64_t)addr;
>         }
>
>
I just spotted that reserve_fill_queue always returns 0.
I understand that xsk_configure expects an errors when not succeeding in
populating this ring.
And for this, it expects a non zero value for this.

How about something like (neither tested nor compiled):

static inline int
reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
{
    struct xsk_ring_prod *fq = &umem->fq;
    void *addrs[reserve_size];
    uint32_t idx;
    int i, ret;

    if (rte_ring_dequeue_bulk(umem->buf_ring, &addrs, reserve_size, NULL)
        != reserve_size) {
        AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
        return -1;
    }

    ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
    if (unlikely(!ret)) {
        AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
        rte_ring_enqueue_bulk(umem->buf_ring, &addrs, reserve_size,
                      NULL);
        return -1;
    }

    for (i = 0; i < reserve_size; i++) {
        __u64 *fq_addr;

        fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
        *fq_addr = (uint64_t)addrs[i];
    }

    xsk_ring_prod__submit(fq, reserve_size);

    return 0;
}



@@ -179,6 +176,9 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>
> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
> +               return 0;
> +
>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>         if (rcvd == 0)
>                 return 0;
>

When xsk_ring_cons__peek() returns 0, we will leak nb_pkts freshly
allocated mbufs.
See below for a suggestion.


@@ -186,9 +186,6 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>
> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
> 0))
> -               return 0;
> -
>         for (i = 0; i < rcvd; i++) {
>                 const struct xdp_desc *desc;
>                 uint64_t addr;
> @@ -211,6 +208,10 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         xsk_ring_cons__release(rx, rcvd);
>
> +       /* free the extra mbufs */
> +       for (; rcvd < nb_pkts; rcvd++)
> +               rte_pktmbuf_free(mbufs[rcvd]);
> +
>

You can move this block after the statistic update...


        /* statistics */
>         rxq->stats.rx_pkts += (rcvd - dropped);
>         rxq->stats.rx_bytes += rx_bytes;
>

... then define a out: label.
And those mbufs are still clean and coming from a single mempool, we can
put them back as a single bulk.
Something like (again, untested):

out:
    if (count != nb_pkts) {
        rte_mempool_put_bulk(rxq->mb_pool, &mbufs[count],
                     nb_pkts - count);
    }

    return count;
}

And you would jump to this label when xsk_ring_cons__peek() == 0.
What do you think ?



@@ -261,55 +262,56 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct xsk_umem_info *umem = txq->pair->umem;
>         struct rte_mbuf *mbuf;
>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
> +       struct rte_mbuf *valid_bufs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long tx_bytes = 0;
> -       int i, valid = 0;
> +       int i;
> +       uint16_t nb_valid = 0;
>         uint32_t idx_tx;
> +       uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE -
> ETH_AF_XDP_DATA_HEADROOM;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>
>         pull_umem_cq(umem, nb_pkts);
>
> -       nb_pkts = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
> -                                       nb_pkts, NULL);
> -       if (nb_pkts == 0)
> +       for (i = 0; i < nb_pkts; i++) {
> +               if (bufs[i]->pkt_len <= buf_len)
> +                       valid_bufs[nb_valid++] = bufs[i];
> +               else
> +                       rte_pktmbuf_free(bufs[i]);
> +       }
> +
> +       nb_valid = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
> +                                       nb_valid, NULL);
> +       if (nb_valid == 0)
>                 return 0;
>
>
You can't return 0.
You have stolen buffers from the caller with the previous check on pktlen.
When the application resends this bulk or frees the whole bulk, we will
have mbuf reuse bugs.

Thinking about this, why would this happen ?
This limitation should be handled by properly reporting the mtu.
The application would then know it can't send those too big mbufs.


If I missed something else and/or if you still don't trust the application,
I think the tx burst function should go like as described below.

The first thing to do is check the mbufs length.
At the first invalid length, you break from the loop at index i (i is the
invalid buffer index).
Then dequeue i - 1 buffers from buf_ring.
Reserve i - 1 slots in tx.

And return i buffers have been sent (plus a tx error stat += 1).

You need to carefully take into account each step and free the buffer to
buf_ring when relevant and free the mbufs properly.


-       if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
> +       if (xsk_ring_prod__reserve(&txq->tx, nb_valid, &idx_tx) !=
> nb_valid) {
>                 kick_tx(txq);
> -               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_valid,
> NULL);
>                 return 0;
>         }
>
> -       for (i = 0; i < nb_pkts; i++) {
> +       for (i = 0; i < nb_valid; i++) {
>                 struct xdp_desc *desc;
>                 void *pkt;
> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
> -                                       - ETH_AF_XDP_DATA_HEADROOM;
>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
> -               mbuf = bufs[i];
> -               if (mbuf->pkt_len <= buf_len) {
> -                       desc->addr = (uint64_t)addrs[valid];
> -                       desc->len = mbuf->pkt_len;
> -                       pkt = xsk_umem__get_data(umem->mz->addr,
> -                                                desc->addr);
> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> -                              desc->len);
> -                       valid++;
> -                       tx_bytes += mbuf->pkt_len;
> -               }
> +               mbuf = valid_bufs[i];
> +               desc->addr = (uint64_t)addrs[i];
> +               desc->len = mbuf->pkt_len;
> +               pkt = xsk_umem__get_data(umem->mz->addr,
> +                                        desc->addr);
> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> +                          desc->len);
> +               tx_bytes += mbuf->pkt_len;
>                 rte_pktmbuf_free(mbuf);
>         }
>
> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
> +       xsk_ring_prod__submit(&txq->tx, nb_valid);
>
>         kick_tx(txq);
>
> -       if (valid < nb_pkts)
> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
> -                                nb_pkts - valid, NULL);
> -
> -       txq->stats.err_pkts += nb_pkts - valid;
> -       txq->stats.tx_pkts += valid;
> +       txq->stats.err_pkts += nb_pkts - nb_valid;
> +       txq->stats.tx_pkts += nb_valid;
>         txq->stats.tx_bytes += tx_bytes;
>
>         return nb_pkts;
> --
> 2.17.1
>
>

-- 
David Marchand

Re: [dpdk-dev] [PATCH 2/2] net/af_xdp: make reserve/submit peek/release consistent

[David Marchand]

On Fri, Apr 12, 2019 at 4:54 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> As David pointed out, if we reserve N slots, but only submit n slots,
> we would end up with an incorrect opinion of the number of available slots
> later, we also would get wrong idx when we call xsk_ring_prod__reserve next
> time. It also applies to xsk_ring_cons__peek()/xsk_ring_cons__release().
>
> This patch ensures that both reserve/submit and peek/release are
> consistent.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Reported-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 80 +++++++++++++++--------------
>  1 file changed, 41 insertions(+), 39 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..76a6a8331 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -138,22 +138,19 @@ reserve_fill_queue(struct xsk_umem_info *umem, int
> reserve_size)
>  {
>         struct xsk_ring_prod *fq = &umem->fq;
>         uint32_t idx;
> -       int i, ret;
> -
> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
> -       if (unlikely(!ret)) {
> -               AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
> -               return ret;
> -       }
> +       int i;
>
>         for (i = 0; i < reserve_size; i++) {
>                 __u64 *fq_addr;
>                 void *addr = NULL;
>                 if (rte_ring_dequeue(umem->buf_ring, &addr)) {
> -                       i--;
>                         break;
>                 }
> -               fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
> +               if (unlikely(!xsk_ring_prod__reserve(fq, 1, &idx))) {
> +                       AF_XDP_LOG(WARNING, "Failed to reserve 1 fq
> desc.\n");
> +                       break;
> +               }
> +               fq_addr = xsk_ring_prod__fill_addr(fq, idx);
>                 *fq_addr = (uint64_t)addr;
>         }
>
>
I just spotted that reserve_fill_queue always returns 0.
I understand that xsk_configure expects an errors when not succeeding in
populating this ring.
And for this, it expects a non zero value for this.

How about something like (neither tested nor compiled):

static inline int
reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
{
    struct xsk_ring_prod *fq = &umem->fq;
    void *addrs[reserve_size];
    uint32_t idx;
    int i, ret;

    if (rte_ring_dequeue_bulk(umem->buf_ring, &addrs, reserve_size, NULL)
        != reserve_size) {
        AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
        return -1;
    }

    ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
    if (unlikely(!ret)) {
        AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
        rte_ring_enqueue_bulk(umem->buf_ring, &addrs, reserve_size,
                      NULL);
        return -1;
    }

    for (i = 0; i < reserve_size; i++) {
        __u64 *fq_addr;

        fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
        *fq_addr = (uint64_t)addrs[i];
    }

    xsk_ring_prod__submit(fq, reserve_size);

    return 0;
}



@@ -179,6 +176,9 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>
> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
> +               return 0;
> +
>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>         if (rcvd == 0)
>                 return 0;
>

When xsk_ring_cons__peek() returns 0, we will leak nb_pkts freshly
allocated mbufs.
See below for a suggestion.


@@ -186,9 +186,6 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>
> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
> 0))
> -               return 0;
> -
>         for (i = 0; i < rcvd; i++) {
>                 const struct xdp_desc *desc;
>                 uint64_t addr;
> @@ -211,6 +208,10 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         xsk_ring_cons__release(rx, rcvd);
>
> +       /* free the extra mbufs */
> +       for (; rcvd < nb_pkts; rcvd++)
> +               rte_pktmbuf_free(mbufs[rcvd]);
> +
>

You can move this block after the statistic update...


        /* statistics */
>         rxq->stats.rx_pkts += (rcvd - dropped);
>         rxq->stats.rx_bytes += rx_bytes;
>

... then define a out: label.
And those mbufs are still clean and coming from a single mempool, we can
put them back as a single bulk.
Something like (again, untested):

out:
    if (count != nb_pkts) {
        rte_mempool_put_bulk(rxq->mb_pool, &mbufs[count],
                     nb_pkts - count);
    }

    return count;
}

And you would jump to this label when xsk_ring_cons__peek() == 0.
What do you think ?



@@ -261,55 +262,56 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct xsk_umem_info *umem = txq->pair->umem;
>         struct rte_mbuf *mbuf;
>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
> +       struct rte_mbuf *valid_bufs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long tx_bytes = 0;
> -       int i, valid = 0;
> +       int i;
> +       uint16_t nb_valid = 0;
>         uint32_t idx_tx;
> +       uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE -
> ETH_AF_XDP_DATA_HEADROOM;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>
>         pull_umem_cq(umem, nb_pkts);
>
> -       nb_pkts = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
> -                                       nb_pkts, NULL);
> -       if (nb_pkts == 0)
> +       for (i = 0; i < nb_pkts; i++) {
> +               if (bufs[i]->pkt_len <= buf_len)
> +                       valid_bufs[nb_valid++] = bufs[i];
> +               else
> +                       rte_pktmbuf_free(bufs[i]);
> +       }
> +
> +       nb_valid = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
> +                                       nb_valid, NULL);
> +       if (nb_valid == 0)
>                 return 0;
>
>
You can't return 0.
You have stolen buffers from the caller with the previous check on pktlen.
When the application resends this bulk or frees the whole bulk, we will
have mbuf reuse bugs.

Thinking about this, why would this happen ?
This limitation should be handled by properly reporting the mtu.
The application would then know it can't send those too big mbufs.


If I missed something else and/or if you still don't trust the application,
I think the tx burst function should go like as described below.

The first thing to do is check the mbufs length.
At the first invalid length, you break from the loop at index i (i is the
invalid buffer index).
Then dequeue i - 1 buffers from buf_ring.
Reserve i - 1 slots in tx.

And return i buffers have been sent (plus a tx error stat += 1).

You need to carefully take into account each step and free the buffer to
buf_ring when relevant and free the mbufs properly.


-       if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
> +       if (xsk_ring_prod__reserve(&txq->tx, nb_valid, &idx_tx) !=
> nb_valid) {
>                 kick_tx(txq);
> -               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_valid,
> NULL);
>                 return 0;
>         }
>
> -       for (i = 0; i < nb_pkts; i++) {
> +       for (i = 0; i < nb_valid; i++) {
>                 struct xdp_desc *desc;
>                 void *pkt;
> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
> -                                       - ETH_AF_XDP_DATA_HEADROOM;
>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
> -               mbuf = bufs[i];
> -               if (mbuf->pkt_len <= buf_len) {
> -                       desc->addr = (uint64_t)addrs[valid];
> -                       desc->len = mbuf->pkt_len;
> -                       pkt = xsk_umem__get_data(umem->mz->addr,
> -                                                desc->addr);
> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> -                              desc->len);
> -                       valid++;
> -                       tx_bytes += mbuf->pkt_len;
> -               }
> +               mbuf = valid_bufs[i];
> +               desc->addr = (uint64_t)addrs[i];
> +               desc->len = mbuf->pkt_len;
> +               pkt = xsk_umem__get_data(umem->mz->addr,
> +                                        desc->addr);
> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> +                          desc->len);
> +               tx_bytes += mbuf->pkt_len;
>                 rte_pktmbuf_free(mbuf);
>         }
>
> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
> +       xsk_ring_prod__submit(&txq->tx, nb_valid);
>
>         kick_tx(txq);
>
> -       if (valid < nb_pkts)
> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
> -                                nb_pkts - valid, NULL);
> -
> -       txq->stats.err_pkts += nb_pkts - valid;
> -       txq->stats.tx_pkts += valid;
> +       txq->stats.err_pkts += nb_pkts - nb_valid;
> +       txq->stats.tx_pkts += nb_valid;
>         txq->stats.tx_bytes += tx_bytes;
>
>         return nb_pkts;
> --
> 2.17.1
>
>

-- 
David Marchand

Re: [dpdk-dev] [PATCH 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Ye Xiaolong]

Hi, David

Thanks for you detailed review comment. 

On 04/15, David Marchand wrote:
>On Fri, Apr 12, 2019 at 4:54 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> As David pointed out, if we reserve N slots, but only submit n slots,
>> we would end up with an incorrect opinion of the number of available slots
>> later, we also would get wrong idx when we call xsk_ring_prod__reserve next
>> time. It also applies to xsk_ring_cons__peek()/xsk_ring_cons__release().
>>
>> This patch ensures that both reserve/submit and peek/release are
>> consistent.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Reported-by: David Marchand <david.marchand@redhat.com>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 80 +++++++++++++++--------------
>>  1 file changed, 41 insertions(+), 39 deletions(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..76a6a8331 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -138,22 +138,19 @@ reserve_fill_queue(struct xsk_umem_info *umem, int
>> reserve_size)
>>  {
>>         struct xsk_ring_prod *fq = &umem->fq;
>>         uint32_t idx;
>> -       int i, ret;
>> -
>> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
>> -       if (unlikely(!ret)) {
>> -               AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
>> -               return ret;
>> -       }
>> +       int i;
>>
>>         for (i = 0; i < reserve_size; i++) {
>>                 __u64 *fq_addr;
>>                 void *addr = NULL;
>>                 if (rte_ring_dequeue(umem->buf_ring, &addr)) {
>> -                       i--;
>>                         break;
>>                 }
>> -               fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
>> +               if (unlikely(!xsk_ring_prod__reserve(fq, 1, &idx))) {
>> +                       AF_XDP_LOG(WARNING, "Failed to reserve 1 fq
>> desc.\n");
>> +                       break;
>> +               }
>> +               fq_addr = xsk_ring_prod__fill_addr(fq, idx);
>>                 *fq_addr = (uint64_t)addr;
>>         }
>>
>>
>I just spotted that reserve_fill_queue always returns 0.
>I understand that xsk_configure expects an errors when not succeeding in
>populating this ring.
>And for this, it expects a non zero value for this.

You are right, reserve_fill_queue does need retrun a non zero value when
it fails to populate the ring.

>
>How about something like (neither tested nor compiled):
>
>static inline int
>reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
>{
>    struct xsk_ring_prod *fq = &umem->fq;
>    void *addrs[reserve_size];
>    uint32_t idx;
>    int i, ret;
>
>    if (rte_ring_dequeue_bulk(umem->buf_ring, &addrs, reserve_size, NULL)
>        != reserve_size) {
>        AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
>        return -1;
>    }
>
>    ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
>    if (unlikely(!ret)) {
>        AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
>        rte_ring_enqueue_bulk(umem->buf_ring, &addrs, reserve_size,
>                      NULL);
>        return -1;
>    }
>
>    for (i = 0; i < reserve_size; i++) {
>        __u64 *fq_addr;
>
>        fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
>        *fq_addr = (uint64_t)addrs[i];
>    }
>
>    xsk_ring_prod__submit(fq, reserve_size);
>
>    return 0;
>}

Sounds better, I'll adopt it in my new version.

>
>
>
>@@ -179,6 +176,9 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>>
>> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
>> != 0))
>> +               return 0;
>> +
>>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>>         if (rcvd == 0)
>>                 return 0;
>>
>
>When xsk_ring_cons__peek() returns 0, we will leak nb_pkts freshly
>allocated mbufs.
>See below for a suggestion.
>
>
>@@ -186,9 +186,6 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>>
>> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
>> 0))
>> -               return 0;
>> -
>>         for (i = 0; i < rcvd; i++) {
>>                 const struct xdp_desc *desc;
>>                 uint64_t addr;
>> @@ -211,6 +208,10 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         xsk_ring_cons__release(rx, rcvd);
>>
>> +       /* free the extra mbufs */
>> +       for (; rcvd < nb_pkts; rcvd++)
>> +               rte_pktmbuf_free(mbufs[rcvd]);
>> +
>>
>
>You can move this block after the statistic update...
>
>
>        /* statistics */
>>         rxq->stats.rx_pkts += (rcvd - dropped);
>>         rxq->stats.rx_bytes += rx_bytes;
>>
>
>... then define a out: label.
>And those mbufs are still clean and coming from a single mempool, we can
>put them back as a single bulk.
>Something like (again, untested):
>
>out:
>    if (count != nb_pkts) {
>        rte_mempool_put_bulk(rxq->mb_pool, &mbufs[count],
>                     nb_pkts - count);
>    }
>
>    return count;
>}
>
>And you would jump to this label when xsk_ring_cons__peek() == 0.
>What do you think ?

I think these are all sensible suggestions, will do.

>
>
>
>@@ -261,55 +262,56 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         struct xsk_umem_info *umem = txq->pair->umem;
>>         struct rte_mbuf *mbuf;
>>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>> +       struct rte_mbuf *valid_bufs[ETH_AF_XDP_TX_BATCH_SIZE];
>>         unsigned long tx_bytes = 0;
>> -       int i, valid = 0;
>> +       int i;
>> +       uint16_t nb_valid = 0;
>>         uint32_t idx_tx;
>> +       uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE -
>> ETH_AF_XDP_DATA_HEADROOM;
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>>
>>         pull_umem_cq(umem, nb_pkts);
>>
>> -       nb_pkts = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
>> -                                       nb_pkts, NULL);
>> -       if (nb_pkts == 0)
>> +       for (i = 0; i < nb_pkts; i++) {
>> +               if (bufs[i]->pkt_len <= buf_len)
>> +                       valid_bufs[nb_valid++] = bufs[i];
>> +               else
>> +                       rte_pktmbuf_free(bufs[i]);
>> +       }
>> +
>> +       nb_valid = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
>> +                                       nb_valid, NULL);
>> +       if (nb_valid == 0)
>>                 return 0;
>>
>>
>You can't return 0.
>You have stolen buffers from the caller with the previous check on pktlen.
>When the application resends this bulk or frees the whole bulk, we will
>have mbuf reuse bugs.
>
>Thinking about this, why would this happen ?
>This limitation should be handled by properly reporting the mtu.
>The application would then know it can't send those too big mbufs.
>

I think we can rely on mtu to handle the limitation, and assume that in this
tx function, all pktlen are valid. Will change in next version.

Thanks,
Xiaolong
>
>If I missed something else and/or if you still don't trust the application,
>I think the tx burst function should go like as described below.
>
>The first thing to do is check the mbufs length.
>At the first invalid length, you break from the loop at index i (i is the
>invalid buffer index).
>Then dequeue i - 1 buffers from buf_ring.
>Reserve i - 1 slots in tx.
>
>And return i buffers have been sent (plus a tx error stat += 1).
>
>You need to carefully take into account each step and free the buffer to
>buf_ring when relevant and free the mbufs properly.
>
>
>-       if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
>> +       if (xsk_ring_prod__reserve(&txq->tx, nb_valid, &idx_tx) !=
>> nb_valid) {
>>                 kick_tx(txq);
>> -               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
>> NULL);
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_valid,
>> NULL);
>>                 return 0;
>>         }
>>
>> -       for (i = 0; i < nb_pkts; i++) {
>> +       for (i = 0; i < nb_valid; i++) {
>>                 struct xdp_desc *desc;
>>                 void *pkt;
>> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
>> -                                       - ETH_AF_XDP_DATA_HEADROOM;
>>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>> -               mbuf = bufs[i];
>> -               if (mbuf->pkt_len <= buf_len) {
>> -                       desc->addr = (uint64_t)addrs[valid];
>> -                       desc->len = mbuf->pkt_len;
>> -                       pkt = xsk_umem__get_data(umem->mz->addr,
>> -                                                desc->addr);
>> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> -                              desc->len);
>> -                       valid++;
>> -                       tx_bytes += mbuf->pkt_len;
>> -               }
>> +               mbuf = valid_bufs[i];
>> +               desc->addr = (uint64_t)addrs[i];
>> +               desc->len = mbuf->pkt_len;
>> +               pkt = xsk_umem__get_data(umem->mz->addr,
>> +                                        desc->addr);
>> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> +                          desc->len);
>> +               tx_bytes += mbuf->pkt_len;
>>                 rte_pktmbuf_free(mbuf);
>>         }
>>
>> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
>> +       xsk_ring_prod__submit(&txq->tx, nb_valid);
>>
>>         kick_tx(txq);
>>
>> -       if (valid < nb_pkts)
>> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
>> -                                nb_pkts - valid, NULL);
>> -
>> -       txq->stats.err_pkts += nb_pkts - valid;
>> -       txq->stats.tx_pkts += valid;
>> +       txq->stats.err_pkts += nb_pkts - nb_valid;
>> +       txq->stats.tx_pkts += nb_valid;
>>         txq->stats.tx_bytes += tx_bytes;
>>
>>         return nb_pkts;
>> --
>> 2.17.1
>>
>>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Ye Xiaolong]

Hi, David

Thanks for you detailed review comment. 

On 04/15, David Marchand wrote:
>On Fri, Apr 12, 2019 at 4:54 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> As David pointed out, if we reserve N slots, but only submit n slots,
>> we would end up with an incorrect opinion of the number of available slots
>> later, we also would get wrong idx when we call xsk_ring_prod__reserve next
>> time. It also applies to xsk_ring_cons__peek()/xsk_ring_cons__release().
>>
>> This patch ensures that both reserve/submit and peek/release are
>> consistent.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Reported-by: David Marchand <david.marchand@redhat.com>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 80 +++++++++++++++--------------
>>  1 file changed, 41 insertions(+), 39 deletions(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..76a6a8331 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -138,22 +138,19 @@ reserve_fill_queue(struct xsk_umem_info *umem, int
>> reserve_size)
>>  {
>>         struct xsk_ring_prod *fq = &umem->fq;
>>         uint32_t idx;
>> -       int i, ret;
>> -
>> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
>> -       if (unlikely(!ret)) {
>> -               AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
>> -               return ret;
>> -       }
>> +       int i;
>>
>>         for (i = 0; i < reserve_size; i++) {
>>                 __u64 *fq_addr;
>>                 void *addr = NULL;
>>                 if (rte_ring_dequeue(umem->buf_ring, &addr)) {
>> -                       i--;
>>                         break;
>>                 }
>> -               fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
>> +               if (unlikely(!xsk_ring_prod__reserve(fq, 1, &idx))) {
>> +                       AF_XDP_LOG(WARNING, "Failed to reserve 1 fq
>> desc.\n");
>> +                       break;
>> +               }
>> +               fq_addr = xsk_ring_prod__fill_addr(fq, idx);
>>                 *fq_addr = (uint64_t)addr;
>>         }
>>
>>
>I just spotted that reserve_fill_queue always returns 0.
>I understand that xsk_configure expects an errors when not succeeding in
>populating this ring.
>And for this, it expects a non zero value for this.

You are right, reserve_fill_queue does need retrun a non zero value when
it fails to populate the ring.

>
>How about something like (neither tested nor compiled):
>
>static inline int
>reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
>{
>    struct xsk_ring_prod *fq = &umem->fq;
>    void *addrs[reserve_size];
>    uint32_t idx;
>    int i, ret;
>
>    if (rte_ring_dequeue_bulk(umem->buf_ring, &addrs, reserve_size, NULL)
>        != reserve_size) {
>        AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
>        return -1;
>    }
>
>    ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
>    if (unlikely(!ret)) {
>        AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
>        rte_ring_enqueue_bulk(umem->buf_ring, &addrs, reserve_size,
>                      NULL);
>        return -1;
>    }
>
>    for (i = 0; i < reserve_size; i++) {
>        __u64 *fq_addr;
>
>        fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
>        *fq_addr = (uint64_t)addrs[i];
>    }
>
>    xsk_ring_prod__submit(fq, reserve_size);
>
>    return 0;
>}

Sounds better, I'll adopt it in my new version.

>
>
>
>@@ -179,6 +176,9 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>>
>> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
>> != 0))
>> +               return 0;
>> +
>>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>>         if (rcvd == 0)
>>                 return 0;
>>
>
>When xsk_ring_cons__peek() returns 0, we will leak nb_pkts freshly
>allocated mbufs.
>See below for a suggestion.
>
>
>@@ -186,9 +186,6 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>>
>> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
>> 0))
>> -               return 0;
>> -
>>         for (i = 0; i < rcvd; i++) {
>>                 const struct xdp_desc *desc;
>>                 uint64_t addr;
>> @@ -211,6 +208,10 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         xsk_ring_cons__release(rx, rcvd);
>>
>> +       /* free the extra mbufs */
>> +       for (; rcvd < nb_pkts; rcvd++)
>> +               rte_pktmbuf_free(mbufs[rcvd]);
>> +
>>
>
>You can move this block after the statistic update...
>
>
>        /* statistics */
>>         rxq->stats.rx_pkts += (rcvd - dropped);
>>         rxq->stats.rx_bytes += rx_bytes;
>>
>
>... then define a out: label.
>And those mbufs are still clean and coming from a single mempool, we can
>put them back as a single bulk.
>Something like (again, untested):
>
>out:
>    if (count != nb_pkts) {
>        rte_mempool_put_bulk(rxq->mb_pool, &mbufs[count],
>                     nb_pkts - count);
>    }
>
>    return count;
>}
>
>And you would jump to this label when xsk_ring_cons__peek() == 0.
>What do you think ?

I think these are all sensible suggestions, will do.

>
>
>
>@@ -261,55 +262,56 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         struct xsk_umem_info *umem = txq->pair->umem;
>>         struct rte_mbuf *mbuf;
>>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>> +       struct rte_mbuf *valid_bufs[ETH_AF_XDP_TX_BATCH_SIZE];
>>         unsigned long tx_bytes = 0;
>> -       int i, valid = 0;
>> +       int i;
>> +       uint16_t nb_valid = 0;
>>         uint32_t idx_tx;
>> +       uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE -
>> ETH_AF_XDP_DATA_HEADROOM;
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>>
>>         pull_umem_cq(umem, nb_pkts);
>>
>> -       nb_pkts = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
>> -                                       nb_pkts, NULL);
>> -       if (nb_pkts == 0)
>> +       for (i = 0; i < nb_pkts; i++) {
>> +               if (bufs[i]->pkt_len <= buf_len)
>> +                       valid_bufs[nb_valid++] = bufs[i];
>> +               else
>> +                       rte_pktmbuf_free(bufs[i]);
>> +       }
>> +
>> +       nb_valid = rte_ring_dequeue_bulk(umem->buf_ring, addrs,
>> +                                       nb_valid, NULL);
>> +       if (nb_valid == 0)
>>                 return 0;
>>
>>
>You can't return 0.
>You have stolen buffers from the caller with the previous check on pktlen.
>When the application resends this bulk or frees the whole bulk, we will
>have mbuf reuse bugs.
>
>Thinking about this, why would this happen ?
>This limitation should be handled by properly reporting the mtu.
>The application would then know it can't send those too big mbufs.
>

I think we can rely on mtu to handle the limitation, and assume that in this
tx function, all pktlen are valid. Will change in next version.

Thanks,
Xiaolong
>
>If I missed something else and/or if you still don't trust the application,
>I think the tx burst function should go like as described below.
>
>The first thing to do is check the mbufs length.
>At the first invalid length, you break from the loop at index i (i is the
>invalid buffer index).
>Then dequeue i - 1 buffers from buf_ring.
>Reserve i - 1 slots in tx.
>
>And return i buffers have been sent (plus a tx error stat += 1).
>
>You need to carefully take into account each step and free the buffer to
>buf_ring when relevant and free the mbufs properly.
>
>
>-       if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
>> +       if (xsk_ring_prod__reserve(&txq->tx, nb_valid, &idx_tx) !=
>> nb_valid) {
>>                 kick_tx(txq);
>> -               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
>> NULL);
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_valid,
>> NULL);
>>                 return 0;
>>         }
>>
>> -       for (i = 0; i < nb_pkts; i++) {
>> +       for (i = 0; i < nb_valid; i++) {
>>                 struct xdp_desc *desc;
>>                 void *pkt;
>> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
>> -                                       - ETH_AF_XDP_DATA_HEADROOM;
>>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>> -               mbuf = bufs[i];
>> -               if (mbuf->pkt_len <= buf_len) {
>> -                       desc->addr = (uint64_t)addrs[valid];
>> -                       desc->len = mbuf->pkt_len;
>> -                       pkt = xsk_umem__get_data(umem->mz->addr,
>> -                                                desc->addr);
>> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> -                              desc->len);
>> -                       valid++;
>> -                       tx_bytes += mbuf->pkt_len;
>> -               }
>> +               mbuf = valid_bufs[i];
>> +               desc->addr = (uint64_t)addrs[i];
>> +               desc->len = mbuf->pkt_len;
>> +               pkt = xsk_umem__get_data(umem->mz->addr,
>> +                                        desc->addr);
>> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> +                          desc->len);
>> +               tx_bytes += mbuf->pkt_len;
>>                 rte_pktmbuf_free(mbuf);
>>         }
>>
>> -       xsk_ring_prod__submit(&txq->tx, nb_pkts);
>> +       xsk_ring_prod__submit(&txq->tx, nb_valid);
>>
>>         kick_tx(txq);
>>
>> -       if (valid < nb_pkts)
>> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
>> -                                nb_pkts - valid, NULL);
>> -
>> -       txq->stats.err_pkts += nb_pkts - valid;
>> -       txq->stats.tx_pkts += valid;
>> +       txq->stats.err_pkts += nb_pkts - nb_valid;
>> +       txq->stats.tx_pkts += nb_valid;
>>         txq->stats.tx_bytes += tx_bytes;
>>
>>         return nb_pkts;
>> --
>> 2.17.1
>>
>>
>
>-- 
>David Marchand

[dpdk-dev] [PATCH v2 0/2] some fixes

[Xiaolong Ye]

This patchset provides some fixes to af_xdp pmd, at first, I just added
a simple error handling when Tx queue allocation fails, then David
suggested a better way to do it and pointed out the inconsistent issue
of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
queue), the second patch addressed this.

v2 changes:

- adopt David's suggestion to refactor the code

Xiaolong Ye (2):
  net/af_xdp: enqueue buf ring when allocate Tx queue fails
  net/af_xdp: make reserve/submit peek/release consistent

 drivers/net/af_xdp/rte_eth_af_xdp.c | 78 +++++++++++++++--------------
 1 file changed, 40 insertions(+), 38 deletions(-)

-- 
2.17.1

[dpdk-dev] [PATCH v2 0/2] some fixes

[Xiaolong Ye]

This patchset provides some fixes to af_xdp pmd, at first, I just added
a simple error handling when Tx queue allocation fails, then David
suggested a better way to do it and pointed out the inconsistent issue
of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
queue), the second patch addressed this.

v2 changes:

- adopt David's suggestion to refactor the code

Xiaolong Ye (2):
  net/af_xdp: enqueue buf ring when allocate Tx queue fails
  net/af_xdp: make reserve/submit peek/release consistent

 drivers/net/af_xdp/rte_eth_af_xdp.c | 78 +++++++++++++++--------------
 1 file changed, 40 insertions(+), 38 deletions(-)

-- 
2.17.1

[dpdk-dev] [PATCH v2 1/2] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH v2 1/2] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH v2 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots for Tx, but only submit n
slots, we would end up with an incorrect opinion of the number of available
slots later, we also would get wrong idx when we call
xsk_ring_prod__reserve next time. It also applies to
xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Suggested-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 77 +++++++++++++++--------------
 1 file changed, 39 insertions(+), 38 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..b00cd6e03 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
 };
 
 static inline int
-reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
+reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
+	void *addrs[reserve_size];
 	uint32_t idx;
-	int i, ret;
+	uint16_t i;
+
+	if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size, NULL)
+		    != reserve_size) {
+		AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
+		return -1;
+	}
 
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
+	if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
 		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs,
+				reserve_size, NULL);
+		return -1;
 	}
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
-		void *addr = NULL;
-		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
-			break;
-		}
+
 		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
-		*fq_addr = (uint64_t)addr;
+		*fq_addr = (uint64_t)addrs[i];
 	}
 
-	xsk_ring_prod__submit(fq, i);
+	xsk_ring_prod__submit(fq, reserve_size);
 
 	return 0;
 }
@@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
-	uint16_t count = 0;
 	int rcvd, i;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
-		return 0;
+		goto out;
 
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_pkt_len(mbufs[i]) = len;
 		rte_pktmbuf_data_len(mbufs[i]) = len;
 		rx_bytes += len;
-		bufs[count++] = mbufs[i];
+		bufs[i] = mbufs[i];
 
 		rte_ring_enqueue(umem->buf_ring, (void *)addr);
 	}
@@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
 
-	return count;
+ out:
+	if (rcvd != nb_pkts)
+		rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
+				     nb_pkts - rcvd);
+
+	return rcvd;
 }
 
 static void
@@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
 	uint32_t idx_tx;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
@@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	for (i = 0; i < nb_pkts; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
+
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
 		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
+
 		rte_pktmbuf_free(mbuf);
 	}
 
@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.tx_pkts += nb_pkts;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

[dpdk-dev] [PATCH v2 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots for Tx, but only submit n
slots, we would end up with an incorrect opinion of the number of available
slots later, we also would get wrong idx when we call
xsk_ring_prod__reserve next time. It also applies to
xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Suggested-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 77 +++++++++++++++--------------
 1 file changed, 39 insertions(+), 38 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..b00cd6e03 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
 };
 
 static inline int
-reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
+reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
+	void *addrs[reserve_size];
 	uint32_t idx;
-	int i, ret;
+	uint16_t i;
+
+	if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size, NULL)
+		    != reserve_size) {
+		AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
+		return -1;
+	}
 
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
+	if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
 		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs,
+				reserve_size, NULL);
+		return -1;
 	}
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
-		void *addr = NULL;
-		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
-			break;
-		}
+
 		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
-		*fq_addr = (uint64_t)addr;
+		*fq_addr = (uint64_t)addrs[i];
 	}
 
-	xsk_ring_prod__submit(fq, i);
+	xsk_ring_prod__submit(fq, reserve_size);
 
 	return 0;
 }
@@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
-	uint16_t count = 0;
 	int rcvd, i;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
-		return 0;
+		goto out;
 
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_pkt_len(mbufs[i]) = len;
 		rte_pktmbuf_data_len(mbufs[i]) = len;
 		rx_bytes += len;
-		bufs[count++] = mbufs[i];
+		bufs[i] = mbufs[i];
 
 		rte_ring_enqueue(umem->buf_ring, (void *)addr);
 	}
@@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
 
-	return count;
+ out:
+	if (rcvd != nb_pkts)
+		rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
+				     nb_pkts - rcvd);
+
+	return rcvd;
 }
 
 static void
@@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
 	uint32_t idx_tx;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
@@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	for (i = 0; i < nb_pkts; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
+
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
 		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
+
 		rte_pktmbuf_free(mbuf);
 	}
 
@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.tx_pkts += nb_pkts;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

Re: [dpdk-dev] [PATCH v2 2/2] net/af_xdp: make reserve/submit peek/release consistent

[David Marchand]

On Tue, Apr 16, 2019 at 5:10 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> As David pointed out, if we reserve N slots for Tx, but only submit n
> slots, we would end up with an incorrect opinion of the number of available
> slots later, we also would get wrong idx when we call
> xsk_ring_prod__reserve next time. It also applies to
> xsk_ring_cons__peek()/xsk_ring_cons__release().
>
> This patch ensures that both reserve/submit and peek/release are
> consistent.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Suggested-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 77 +++++++++++++++--------------
>  1 file changed, 39 insertions(+), 38 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..b00cd6e03 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
>  };
>
>  static inline int
> -reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
> +reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
>  {
>         struct xsk_ring_prod *fq = &umem->fq;
> +       void *addrs[reserve_size];
>         uint32_t idx;
> -       int i, ret;
> +       uint16_t i;
> +
> +       if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size,
> NULL)
> +                   != reserve_size) {
> +               AF_XDP_LOG(DEBUG, "Failed to get enough buffers for
> fq.\n");
> +               return -1;
> +       }
>
> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
> -       if (unlikely(!ret)) {
> +       if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
>                 AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
>

Is this situation happening a lot ?
If this is the case, I would prefer see this as a DEBUG message rather than
ERR.


-               return ret;
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs,
> +                               reserve_size, NULL);
> +               return -1;
>         }
>
>         for (i = 0; i < reserve_size; i++) {
>                 __u64 *fq_addr;
> -               void *addr = NULL;
> -               if (rte_ring_dequeue(umem->buf_ring, &addr)) {
> -                       i--;
> -                       break;
> -               }
> +
>                 fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
> -               *fq_addr = (uint64_t)addr;
> +               *fq_addr = (uint64_t)addrs[i];
>         }
>
> -       xsk_ring_prod__submit(fq, i);
> +       xsk_ring_prod__submit(fq, reserve_size);
>
>         return 0;
>  }
> @@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long dropped = 0;
>         unsigned long rx_bytes = 0;
> -       uint16_t count = 0;
>         int rcvd, i;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>

Nothing to do with your patch, but it should be ETH_AF_XDP_ "R"
X_BATCH_SIZE (we have two occurences of this in eth_af_xdp_rx).
I can send a follow up patch after yours.
Or you do it, your choice :-)



> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
> +               return 0;
> +
>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>         if (rcvd == 0)
> -               return 0;
> +               goto out;
>
>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>
> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
> 0))
> -               return 0;
> -
>         for (i = 0; i < rcvd; i++) {
>                 const struct xdp_desc *desc;
>                 uint64_t addr;
> @@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>                 rte_pktmbuf_pkt_len(mbufs[i]) = len;
>                 rte_pktmbuf_data_len(mbufs[i]) = len;
>                 rx_bytes += len;
> -               bufs[count++] = mbufs[i];
> +               bufs[i] = mbufs[i];
>
>                 rte_ring_enqueue(umem->buf_ring, (void *)addr);
>         }
> @@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         rxq->stats.rx_pkts += (rcvd - dropped);
>         rxq->stats.rx_bytes += rx_bytes;
>
> -       return count;
> + out:
>

No space before label.


+       if (rcvd != nb_pkts)
> +               rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
> +                                    nb_pkts - rcvd);
> +
> +       return rcvd;
>  }
>
>  static void
> @@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbuf;
>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long tx_bytes = 0;
> -       int i, valid = 0;
> +       int i;
>         uint32_t idx_tx;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
> @@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         for (i = 0; i < nb_pkts; i++) {
>                 struct xdp_desc *desc;
>                 void *pkt;
> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
> -                                       - ETH_AF_XDP_DATA_HEADROOM;
> +
>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>                 mbuf = bufs[i];
> -               if (mbuf->pkt_len <= buf_len) {
> -                       desc->addr = (uint64_t)addrs[valid];
> -                       desc->len = mbuf->pkt_len;
> -                       pkt = xsk_umem__get_data(umem->mz->addr,
> -                                                desc->addr);
> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> -                              desc->len);
> -                       valid++;
> -                       tx_bytes += mbuf->pkt_len;
> -               }
> +
> +               desc->addr = (uint64_t)addrs[i];
> +               desc->len = mbuf->pkt_len;
> +               pkt = xsk_umem__get_data(umem->mz->addr,
> +                                        desc->addr);
> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> +                          desc->len);
> +               tx_bytes += mbuf->pkt_len;
> +
>                 rte_pktmbuf_free(mbuf);
>         }
>
>
I can see that the buffers in umem can embed ETH_AF_XDP_FRAME_SIZE -
ETH_AF_XDP_DATA_HEADROOM bytes.
And this driver does not support multi segment.
So we are missing a check in eth_dev_mtu_set().
The min_mtu / max_mtu fields are not filled in eth_dev_info().

You should fix this in a preparation patch before this change.


@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         kick_tx(txq);
>
> -       if (valid < nb_pkts)
> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
> -                                nb_pkts - valid, NULL);
> -
> -       txq->stats.err_pkts += nb_pkts - valid;
>

err_pkts stats is not used anymore afaics.


-       txq->stats.tx_pkts += valid;
> +       txq->stats.tx_pkts += nb_pkts;
>         txq->stats.tx_bytes += tx_bytes;
>
>         return nb_pkts;
> --
> 2.17.1
>
>


-- 
David Marchand

Re: [dpdk-dev] [PATCH v2 2/2] net/af_xdp: make reserve/submit peek/release consistent

[David Marchand]

On Tue, Apr 16, 2019 at 5:10 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> As David pointed out, if we reserve N slots for Tx, but only submit n
> slots, we would end up with an incorrect opinion of the number of available
> slots later, we also would get wrong idx when we call
> xsk_ring_prod__reserve next time. It also applies to
> xsk_ring_cons__peek()/xsk_ring_cons__release().
>
> This patch ensures that both reserve/submit and peek/release are
> consistent.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Suggested-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 77 +++++++++++++++--------------
>  1 file changed, 39 insertions(+), 38 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..b00cd6e03 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
>  };
>
>  static inline int
> -reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
> +reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
>  {
>         struct xsk_ring_prod *fq = &umem->fq;
> +       void *addrs[reserve_size];
>         uint32_t idx;
> -       int i, ret;
> +       uint16_t i;
> +
> +       if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size,
> NULL)
> +                   != reserve_size) {
> +               AF_XDP_LOG(DEBUG, "Failed to get enough buffers for
> fq.\n");
> +               return -1;
> +       }
>
> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
> -       if (unlikely(!ret)) {
> +       if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
>                 AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
>

Is this situation happening a lot ?
If this is the case, I would prefer see this as a DEBUG message rather than
ERR.


-               return ret;
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs,
> +                               reserve_size, NULL);
> +               return -1;
>         }
>
>         for (i = 0; i < reserve_size; i++) {
>                 __u64 *fq_addr;
> -               void *addr = NULL;
> -               if (rte_ring_dequeue(umem->buf_ring, &addr)) {
> -                       i--;
> -                       break;
> -               }
> +
>                 fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
> -               *fq_addr = (uint64_t)addr;
> +               *fq_addr = (uint64_t)addrs[i];
>         }
>
> -       xsk_ring_prod__submit(fq, i);
> +       xsk_ring_prod__submit(fq, reserve_size);
>
>         return 0;
>  }
> @@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long dropped = 0;
>         unsigned long rx_bytes = 0;
> -       uint16_t count = 0;
>         int rcvd, i;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>

Nothing to do with your patch, but it should be ETH_AF_XDP_ "R"
X_BATCH_SIZE (we have two occurences of this in eth_af_xdp_rx).
I can send a follow up patch after yours.
Or you do it, your choice :-)



> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
> +               return 0;
> +
>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>         if (rcvd == 0)
> -               return 0;
> +               goto out;
>
>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>
> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
> 0))
> -               return 0;
> -
>         for (i = 0; i < rcvd; i++) {
>                 const struct xdp_desc *desc;
>                 uint64_t addr;
> @@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>                 rte_pktmbuf_pkt_len(mbufs[i]) = len;
>                 rte_pktmbuf_data_len(mbufs[i]) = len;
>                 rx_bytes += len;
> -               bufs[count++] = mbufs[i];
> +               bufs[i] = mbufs[i];
>
>                 rte_ring_enqueue(umem->buf_ring, (void *)addr);
>         }
> @@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         rxq->stats.rx_pkts += (rcvd - dropped);
>         rxq->stats.rx_bytes += rx_bytes;
>
> -       return count;
> + out:
>

No space before label.


+       if (rcvd != nb_pkts)
> +               rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
> +                                    nb_pkts - rcvd);
> +
> +       return rcvd;
>  }
>
>  static void
> @@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbuf;
>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long tx_bytes = 0;
> -       int i, valid = 0;
> +       int i;
>         uint32_t idx_tx;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
> @@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         for (i = 0; i < nb_pkts; i++) {
>                 struct xdp_desc *desc;
>                 void *pkt;
> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
> -                                       - ETH_AF_XDP_DATA_HEADROOM;
> +
>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>                 mbuf = bufs[i];
> -               if (mbuf->pkt_len <= buf_len) {
> -                       desc->addr = (uint64_t)addrs[valid];
> -                       desc->len = mbuf->pkt_len;
> -                       pkt = xsk_umem__get_data(umem->mz->addr,
> -                                                desc->addr);
> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> -                              desc->len);
> -                       valid++;
> -                       tx_bytes += mbuf->pkt_len;
> -               }
> +
> +               desc->addr = (uint64_t)addrs[i];
> +               desc->len = mbuf->pkt_len;
> +               pkt = xsk_umem__get_data(umem->mz->addr,
> +                                        desc->addr);
> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> +                          desc->len);
> +               tx_bytes += mbuf->pkt_len;
> +
>                 rte_pktmbuf_free(mbuf);
>         }
>
>
I can see that the buffers in umem can embed ETH_AF_XDP_FRAME_SIZE -
ETH_AF_XDP_DATA_HEADROOM bytes.
And this driver does not support multi segment.
So we are missing a check in eth_dev_mtu_set().
The min_mtu / max_mtu fields are not filled in eth_dev_info().

You should fix this in a preparation patch before this change.


@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         kick_tx(txq);
>
> -       if (valid < nb_pkts)
> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
> -                                nb_pkts - valid, NULL);
> -
> -       txq->stats.err_pkts += nb_pkts - valid;
>

err_pkts stats is not used anymore afaics.


-       txq->stats.tx_pkts += valid;
> +       txq->stats.tx_pkts += nb_pkts;
>         txq->stats.tx_bytes += tx_bytes;
>
>         return nb_pkts;
> --
> 2.17.1
>
>


-- 
David Marchand

Re: [dpdk-dev] [PATCH v2 1/2] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[David Marchand]

On Tue, Apr 16, 2019 at 5:09 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When it fails to allocate enough slots in Tx queue for transmitting
> packets, we need to return the dequeued addrs to buf ring.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>  1 file changed, 1 insertion(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..5cc643ce2 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
>                 return 0;
>         }
>
> --
> 2.17.1
>
>
Reviewed-by: David Marchand <david.marchand@redhat.com>


-- 
David Marchand

Re: [dpdk-dev] [PATCH v2 1/2] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[David Marchand]

On Tue, Apr 16, 2019 at 5:09 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When it fails to allocate enough slots in Tx queue for transmitting
> packets, we need to return the dequeued addrs to buf ring.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>  1 file changed, 1 insertion(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..5cc643ce2 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
>                 return 0;
>         }
>
> --
> 2.17.1
>
>
Reviewed-by: David Marchand <david.marchand@redhat.com>


-- 
David Marchand

Re: [dpdk-dev] [PATCH v2 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Tue, Apr 16, 2019 at 5:10 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> As David pointed out, if we reserve N slots for Tx, but only submit n
>> slots, we would end up with an incorrect opinion of the number of available
>> slots later, we also would get wrong idx when we call
>> xsk_ring_prod__reserve next time. It also applies to
>> xsk_ring_cons__peek()/xsk_ring_cons__release().
>>
>> This patch ensures that both reserve/submit and peek/release are
>> consistent.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Suggested-by: David Marchand <david.marchand@redhat.com>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 77 +++++++++++++++--------------
>>  1 file changed, 39 insertions(+), 38 deletions(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..b00cd6e03 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
>>  };
>>
>>  static inline int
>> -reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
>> +reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
>>  {
>>         struct xsk_ring_prod *fq = &umem->fq;
>> +       void *addrs[reserve_size];
>>         uint32_t idx;
>> -       int i, ret;
>> +       uint16_t i;
>> +
>> +       if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size,
>> NULL)
>> +                   != reserve_size) {
>> +               AF_XDP_LOG(DEBUG, "Failed to get enough buffers for
>> fq.\n");
>> +               return -1;
>> +       }
>>
>> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
>> -       if (unlikely(!ret)) {
>> +       if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
>>                 AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
>>
>
>Is this situation happening a lot ?
>If this is the case, I would prefer see this as a DEBUG message rather than
>ERR.

Agree DEBUG lable is better here.

>
>
>-               return ret;
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs,
>> +                               reserve_size, NULL);
>> +               return -1;
>>         }
>>
>>         for (i = 0; i < reserve_size; i++) {
>>                 __u64 *fq_addr;
>> -               void *addr = NULL;
>> -               if (rte_ring_dequeue(umem->buf_ring, &addr)) {
>> -                       i--;
>> -                       break;
>> -               }
>> +
>>                 fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
>> -               *fq_addr = (uint64_t)addr;
>> +               *fq_addr = (uint64_t)addrs[i];
>>         }
>>
>> -       xsk_ring_prod__submit(fq, i);
>> +       xsk_ring_prod__submit(fq, reserve_size);
>>
>>         return 0;
>>  }
>> @@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
>>         unsigned long dropped = 0;
>>         unsigned long rx_bytes = 0;
>> -       uint16_t count = 0;
>>         int rcvd, i;
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>>
>
>Nothing to do with your patch, but it should be ETH_AF_XDP_ "R"
>X_BATCH_SIZE (we have two occurences of this in eth_af_xdp_rx).
>I can send a follow up patch after yours.
>Or you do it, your choice :-)

Good catch, I'll submit a separate patch for this.

>
>
>
>> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
>> != 0))
>> +               return 0;
>> +
>>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>>         if (rcvd == 0)
>> -               return 0;
>> +               goto out;
>>
>>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>>
>> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
>> 0))
>> -               return 0;
>> -
>>         for (i = 0; i < rcvd; i++) {
>>                 const struct xdp_desc *desc;
>>                 uint64_t addr;
>> @@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>                 rte_pktmbuf_pkt_len(mbufs[i]) = len;
>>                 rte_pktmbuf_data_len(mbufs[i]) = len;
>>                 rx_bytes += len;
>> -               bufs[count++] = mbufs[i];
>> +               bufs[i] = mbufs[i];
>>
>>                 rte_ring_enqueue(umem->buf_ring, (void *)addr);
>>         }
>> @@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         rxq->stats.rx_pkts += (rcvd - dropped);
>>         rxq->stats.rx_bytes += rx_bytes;
>>
>> -       return count;
>> + out:
>>
>
>No space before label.

Ouch, will remove this unwanted space.

>
>
>+       if (rcvd != nb_pkts)
>> +               rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
>> +                                    nb_pkts - rcvd);
>> +
>> +       return rcvd;
>>  }
>>
>>  static void
>> @@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         struct rte_mbuf *mbuf;
>>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>>         unsigned long tx_bytes = 0;
>> -       int i, valid = 0;
>> +       int i;
>>         uint32_t idx_tx;
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>> @@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         for (i = 0; i < nb_pkts; i++) {
>>                 struct xdp_desc *desc;
>>                 void *pkt;
>> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
>> -                                       - ETH_AF_XDP_DATA_HEADROOM;
>> +
>>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>>                 mbuf = bufs[i];
>> -               if (mbuf->pkt_len <= buf_len) {
>> -                       desc->addr = (uint64_t)addrs[valid];
>> -                       desc->len = mbuf->pkt_len;
>> -                       pkt = xsk_umem__get_data(umem->mz->addr,
>> -                                                desc->addr);
>> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> -                              desc->len);
>> -                       valid++;
>> -                       tx_bytes += mbuf->pkt_len;
>> -               }
>> +
>> +               desc->addr = (uint64_t)addrs[i];
>> +               desc->len = mbuf->pkt_len;
>> +               pkt = xsk_umem__get_data(umem->mz->addr,
>> +                                        desc->addr);
>> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> +                          desc->len);
>> +               tx_bytes += mbuf->pkt_len;
>> +
>>                 rte_pktmbuf_free(mbuf);
>>         }
>>
>>
>I can see that the buffers in umem can embed ETH_AF_XDP_FRAME_SIZE -
>ETH_AF_XDP_DATA_HEADROOM bytes.
>And this driver does not support multi segment.
>So we are missing a check in eth_dev_mtu_set().
>The min_mtu / max_mtu fields are not filled in eth_dev_info().
>
>You should fix this in a preparation patch before this change.

Got it, will submit a preparation patch in next version.

>
>
>@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         kick_tx(txq);
>>
>> -       if (valid < nb_pkts)
>> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
>> -                                nb_pkts - valid, NULL);
>> -
>> -       txq->stats.err_pkts += nb_pkts - valid;
>>
>
>err_pkts stats is not used anymore afaics.

Will delete it in next version.

Thanks,
Xiaolong

>
>
>-       txq->stats.tx_pkts += valid;
>> +       txq->stats.tx_pkts += nb_pkts;
>>         txq->stats.tx_bytes += tx_bytes;
>>
>>         return nb_pkts;
>> --
>> 2.17.1
>>
>>
>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH v2 2/2] net/af_xdp: make reserve/submit peek/release consistent

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Tue, Apr 16, 2019 at 5:10 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> As David pointed out, if we reserve N slots for Tx, but only submit n
>> slots, we would end up with an incorrect opinion of the number of available
>> slots later, we also would get wrong idx when we call
>> xsk_ring_prod__reserve next time. It also applies to
>> xsk_ring_cons__peek()/xsk_ring_cons__release().
>>
>> This patch ensures that both reserve/submit and peek/release are
>> consistent.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Suggested-by: David Marchand <david.marchand@redhat.com>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 77 +++++++++++++++--------------
>>  1 file changed, 39 insertions(+), 38 deletions(-)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..b00cd6e03 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
>>  };
>>
>>  static inline int
>> -reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
>> +reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
>>  {
>>         struct xsk_ring_prod *fq = &umem->fq;
>> +       void *addrs[reserve_size];
>>         uint32_t idx;
>> -       int i, ret;
>> +       uint16_t i;
>> +
>> +       if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size,
>> NULL)
>> +                   != reserve_size) {
>> +               AF_XDP_LOG(DEBUG, "Failed to get enough buffers for
>> fq.\n");
>> +               return -1;
>> +       }
>>
>> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
>> -       if (unlikely(!ret)) {
>> +       if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
>>                 AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
>>
>
>Is this situation happening a lot ?
>If this is the case, I would prefer see this as a DEBUG message rather than
>ERR.

Agree DEBUG lable is better here.

>
>
>-               return ret;
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs,
>> +                               reserve_size, NULL);
>> +               return -1;
>>         }
>>
>>         for (i = 0; i < reserve_size; i++) {
>>                 __u64 *fq_addr;
>> -               void *addr = NULL;
>> -               if (rte_ring_dequeue(umem->buf_ring, &addr)) {
>> -                       i--;
>> -                       break;
>> -               }
>> +
>>                 fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
>> -               *fq_addr = (uint64_t)addr;
>> +               *fq_addr = (uint64_t)addrs[i];
>>         }
>>
>> -       xsk_ring_prod__submit(fq, i);
>> +       xsk_ring_prod__submit(fq, reserve_size);
>>
>>         return 0;
>>  }
>> @@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
>>         unsigned long dropped = 0;
>>         unsigned long rx_bytes = 0;
>> -       uint16_t count = 0;
>>         int rcvd, i;
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>>
>
>Nothing to do with your patch, but it should be ETH_AF_XDP_ "R"
>X_BATCH_SIZE (we have two occurences of this in eth_af_xdp_rx).
>I can send a follow up patch after yours.
>Or you do it, your choice :-)

Good catch, I'll submit a separate patch for this.

>
>
>
>> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
>> != 0))
>> +               return 0;
>> +
>>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>>         if (rcvd == 0)
>> -               return 0;
>> +               goto out;
>>
>>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>>
>> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
>> 0))
>> -               return 0;
>> -
>>         for (i = 0; i < rcvd; i++) {
>>                 const struct xdp_desc *desc;
>>                 uint64_t addr;
>> @@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>                 rte_pktmbuf_pkt_len(mbufs[i]) = len;
>>                 rte_pktmbuf_data_len(mbufs[i]) = len;
>>                 rx_bytes += len;
>> -               bufs[count++] = mbufs[i];
>> +               bufs[i] = mbufs[i];
>>
>>                 rte_ring_enqueue(umem->buf_ring, (void *)addr);
>>         }
>> @@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         rxq->stats.rx_pkts += (rcvd - dropped);
>>         rxq->stats.rx_bytes += rx_bytes;
>>
>> -       return count;
>> + out:
>>
>
>No space before label.

Ouch, will remove this unwanted space.

>
>
>+       if (rcvd != nb_pkts)
>> +               rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
>> +                                    nb_pkts - rcvd);
>> +
>> +       return rcvd;
>>  }
>>
>>  static void
>> @@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         struct rte_mbuf *mbuf;
>>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>>         unsigned long tx_bytes = 0;
>> -       int i, valid = 0;
>> +       int i;
>>         uint32_t idx_tx;
>>
>>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>> @@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>         for (i = 0; i < nb_pkts; i++) {
>>                 struct xdp_desc *desc;
>>                 void *pkt;
>> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
>> -                                       - ETH_AF_XDP_DATA_HEADROOM;
>> +
>>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>>                 mbuf = bufs[i];
>> -               if (mbuf->pkt_len <= buf_len) {
>> -                       desc->addr = (uint64_t)addrs[valid];
>> -                       desc->len = mbuf->pkt_len;
>> -                       pkt = xsk_umem__get_data(umem->mz->addr,
>> -                                                desc->addr);
>> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> -                              desc->len);
>> -                       valid++;
>> -                       tx_bytes += mbuf->pkt_len;
>> -               }
>> +
>> +               desc->addr = (uint64_t)addrs[i];
>> +               desc->len = mbuf->pkt_len;
>> +               pkt = xsk_umem__get_data(umem->mz->addr,
>> +                                        desc->addr);
>> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
>> +                          desc->len);
>> +               tx_bytes += mbuf->pkt_len;
>> +
>>                 rte_pktmbuf_free(mbuf);
>>         }
>>
>>
>I can see that the buffers in umem can embed ETH_AF_XDP_FRAME_SIZE -
>ETH_AF_XDP_DATA_HEADROOM bytes.
>And this driver does not support multi segment.
>So we are missing a check in eth_dev_mtu_set().
>The min_mtu / max_mtu fields are not filled in eth_dev_info().
>
>You should fix this in a preparation patch before this change.

Got it, will submit a preparation patch in next version.

>
>
>@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         kick_tx(txq);
>>
>> -       if (valid < nb_pkts)
>> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
>> -                                nb_pkts - valid, NULL);
>> -
>> -       txq->stats.err_pkts += nb_pkts - valid;
>>
>
>err_pkts stats is not used anymore afaics.

Will delete it in next version.

Thanks,
Xiaolong

>
>
>-       txq->stats.tx_pkts += valid;
>> +       txq->stats.tx_pkts += nb_pkts;
>>         txq->stats.tx_bytes += tx_bytes;
>>
>>         return nb_pkts;
>> --
>> 2.17.1
>>
>>
>
>
>-- 
>David Marchand

[dpdk-dev] [PATCH v3 0/4] some fixes for AF_XDP pmd

[Xiaolong Ye]

This patchset provides some fixes to af_xdp pmd, at first, I just added
a simple error handling when Tx queue allocation fails, then David
suggested a better way to do it and pointed out the inconsistent issue
of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
queue), the third patch addressed this.

v3 changes:

- address David's review comments
- add one patch to specify the mtu range
- add one fix patch for typo

v2 changes:

- adopt David's suggestion to refactor the code


Xiaolong Ye (4):
  net/af_xdp: enqueue buf ring when allocate Tx queue fails
  net/af_xdp: specify minimal and maximal MTU
  net/af_xdp: make reserve/submit peek/release consistent
  net/af_xdp: fix typos in Rx function

 drivers/net/af_xdp/rte_eth_af_xdp.c | 96 +++++++++++++++++------------
 1 file changed, 55 insertions(+), 41 deletions(-)

-- 
2.17.1

[dpdk-dev] [PATCH v3 0/4] some fixes for AF_XDP pmd

[Xiaolong Ye]

This patchset provides some fixes to af_xdp pmd, at first, I just added
a simple error handling when Tx queue allocation fails, then David
suggested a better way to do it and pointed out the inconsistent issue
of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
queue), the third patch addressed this.

v3 changes:

- address David's review comments
- add one patch to specify the mtu range
- add one fix patch for typo

v2 changes:

- adopt David's suggestion to refactor the code


Xiaolong Ye (4):
  net/af_xdp: enqueue buf ring when allocate Tx queue fails
  net/af_xdp: specify minimal and maximal MTU
  net/af_xdp: make reserve/submit peek/release consistent
  net/af_xdp: fix typos in Rx function

 drivers/net/af_xdp/rte_eth_af_xdp.c | 96 +++++++++++++++++------------
 1 file changed, 55 insertions(+), 41 deletions(-)

-- 
2.17.1

[dpdk-dev] [PATCH v3 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH v3 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH v3 2/4] net/af_xdp: specify minimal and maximal MTU

[Xiaolong Ye]

Since AF_XDP pmd doesn't support multi segment, we need to add a valid
check in eth_dev_mtu_set.

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..8430921af 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct rte_eth_dev_info *dev_info)
 	dev_info->max_rx_queues = 1;
 	dev_info->max_tx_queues = 1;
 
+	dev_info->min_mtu = ETHER_MIN_MTU;
+	dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM;
+
 	dev_info->default_rxportconf.nb_queues = 1;
 	dev_info->default_txportconf.nb_queues = 1;
 	dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
@@ -654,6 +657,15 @@ eth_dev_mtu_set(struct rte_eth_dev *dev, uint16_t mtu)
 	int ret;
 	int s;
 
+	if (mtu > ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM ||
+			mtu < ETHER_MIN_MTU) {
+		AF_XDP_LOG(ERR, "Unsupported MTU of %d. "
+			"max mtu: %d, min mtu: %d", mtu,
+			ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM,
+			ETHER_MIN_MTU);
+		return -EINVAL;
+	}
+
 	s = socket(PF_INET, SOCK_DGRAM, 0);
 	if (s < 0)
 		return -EINVAL;
-- 
2.17.1

[dpdk-dev] [PATCH v3 2/4] net/af_xdp: specify minimal and maximal MTU

[Xiaolong Ye]

Since AF_XDP pmd doesn't support multi segment, we need to add a valid
check in eth_dev_mtu_set.

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..8430921af 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct rte_eth_dev_info *dev_info)
 	dev_info->max_rx_queues = 1;
 	dev_info->max_tx_queues = 1;
 
+	dev_info->min_mtu = ETHER_MIN_MTU;
+	dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM;
+
 	dev_info->default_rxportconf.nb_queues = 1;
 	dev_info->default_txportconf.nb_queues = 1;
 	dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
@@ -654,6 +657,15 @@ eth_dev_mtu_set(struct rte_eth_dev *dev, uint16_t mtu)
 	int ret;
 	int s;
 
+	if (mtu > ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM ||
+			mtu < ETHER_MIN_MTU) {
+		AF_XDP_LOG(ERR, "Unsupported MTU of %d. "
+			"max mtu: %d, min mtu: %d", mtu,
+			ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM,
+			ETHER_MIN_MTU);
+		return -EINVAL;
+	}
+
 	s = socket(PF_INET, SOCK_DGRAM, 0);
 	if (s < 0)
 		return -EINVAL;
-- 
2.17.1

[dpdk-dev] [PATCH v3 3/4] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots for Tx, but only submit n
slots, we would end up with an incorrect opinion of the number of available
slots later, we also would get wrong idx when we call
xsk_ring_prod__reserve next time. It also applies to
xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Suggested-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 79 +++++++++++++++--------------
 1 file changed, 40 insertions(+), 39 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 8430921af..817092584 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
 };
 
 static inline int
-reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
+reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
+	void *addrs[reserve_size];
 	uint32_t idx;
-	int i, ret;
+	uint16_t i;
+
+	if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size, NULL)
+		    != reserve_size) {
+		AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
+		return -1;
+	}
 
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
-		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
+	if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
+		AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs,
+				reserve_size, NULL);
+		return -1;
 	}
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
-		void *addr = NULL;
-		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
-			break;
-		}
+
 		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
-		*fq_addr = (uint64_t)addr;
+		*fq_addr = (uint64_t)addrs[i];
 	}
 
-	xsk_ring_prod__submit(fq, i);
+	xsk_ring_prod__submit(fq, reserve_size);
 
 	return 0;
 }
@@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
-	uint16_t count = 0;
 	int rcvd, i;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
-		return 0;
+		goto out;
 
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_pkt_len(mbufs[i]) = len;
 		rte_pktmbuf_data_len(mbufs[i]) = len;
 		rx_bytes += len;
-		bufs[count++] = mbufs[i];
+		bufs[i] = mbufs[i];
 
 		rte_ring_enqueue(umem->buf_ring, (void *)addr);
 	}
@@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
 
-	return count;
+out:
+	if (rcvd != nb_pkts)
+		rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
+				     nb_pkts - rcvd);
+
+	return rcvd;
 }
 
 static void
@@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
 	uint32_t idx_tx;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
@@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	for (i = 0; i < nb_pkts; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
+
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
 		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
+
 		rte_pktmbuf_free(mbuf);
 	}
 
@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.tx_pkts += nb_pkts;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

[dpdk-dev] [PATCH v3 3/4] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots for Tx, but only submit n
slots, we would end up with an incorrect opinion of the number of available
slots later, we also would get wrong idx when we call
xsk_ring_prod__reserve next time. It also applies to
xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Suggested-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 79 +++++++++++++++--------------
 1 file changed, 40 insertions(+), 39 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 8430921af..817092584 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
 };
 
 static inline int
-reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
+reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
+	void *addrs[reserve_size];
 	uint32_t idx;
-	int i, ret;
+	uint16_t i;
+
+	if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size, NULL)
+		    != reserve_size) {
+		AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
+		return -1;
+	}
 
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
-		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
+	if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
+		AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs,
+				reserve_size, NULL);
+		return -1;
 	}
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
-		void *addr = NULL;
-		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
-			break;
-		}
+
 		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
-		*fq_addr = (uint64_t)addr;
+		*fq_addr = (uint64_t)addrs[i];
 	}
 
-	xsk_ring_prod__submit(fq, i);
+	xsk_ring_prod__submit(fq, reserve_size);
 
 	return 0;
 }
@@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
-	uint16_t count = 0;
 	int rcvd, i;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
-		return 0;
+		goto out;
 
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_pkt_len(mbufs[i]) = len;
 		rte_pktmbuf_data_len(mbufs[i]) = len;
 		rx_bytes += len;
-		bufs[count++] = mbufs[i];
+		bufs[i] = mbufs[i];
 
 		rte_ring_enqueue(umem->buf_ring, (void *)addr);
 	}
@@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
 
-	return count;
+out:
+	if (rcvd != nb_pkts)
+		rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
+				     nb_pkts - rcvd);
+
+	return rcvd;
 }
 
 static void
@@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
 	uint32_t idx_tx;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
@@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	for (i = 0; i < nb_pkts; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
+
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
 		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
+
 		rte_pktmbuf_free(mbuf);
 	}
 
@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.tx_pkts += nb_pkts;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

[dpdk-dev] [PATCH v3 4/4] net/af_xdp: fix typos in Rx function

[Xiaolong Ye]

We should use ETH_AF_XDP_RX_BATCH_SIZE in Rx function rather than
ETH_AF_XDP_TX_BATCH_SIZE.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 817092584..ee10df0dc 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -175,12 +175,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct xsk_ring_prod *fq = &umem->fq;
 	uint32_t idx_rx = 0;
 	uint32_t free_thresh = fq->size >> 1;
-	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
+	struct rte_mbuf *mbufs[ETH_AF_XDP_RX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
 	int rcvd, i;
 
-	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
+	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_RX_BATCH_SIZE);
 
 	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
 		return 0;
-- 
2.17.1

[dpdk-dev] [PATCH v3 4/4] net/af_xdp: fix typos in Rx function

[Xiaolong Ye]

We should use ETH_AF_XDP_RX_BATCH_SIZE in Rx function rather than
ETH_AF_XDP_TX_BATCH_SIZE.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 817092584..ee10df0dc 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -175,12 +175,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct xsk_ring_prod *fq = &umem->fq;
 	uint32_t idx_rx = 0;
 	uint32_t free_thresh = fq->size >> 1;
-	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
+	struct rte_mbuf *mbufs[ETH_AF_XDP_RX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
 	int rcvd, i;
 
-	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
+	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_RX_BATCH_SIZE);
 
 	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
 		return 0;
-- 
2.17.1

Re: [dpdk-dev] [PATCH v3 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When it fails to allocate enough slots in Tx queue for transmitting
> packets, we need to return the dequeued addrs to buf ring.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>  1 file changed, 1 insertion(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..5cc643ce2 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
>                 return 0;
>         }
>
> --
> 2.17.1
>

When there is no change, please keep Review tags from previous versions.

Reviewed-by: David Marchand <david.marchand@redhat.com>

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> When it fails to allocate enough slots in Tx queue for transmitting
> packets, we need to return the dequeued addrs to buf ring.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>  1 file changed, 1 insertion(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 007a1c6b4..5cc643ce2 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
> {
>                 kick_tx(txq);
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
> NULL);
>                 return 0;
>         }
>
> --
> 2.17.1
>

When there is no change, please keep Review tags from previous versions.

Reviewed-by: David Marchand <david.marchand@redhat.com>

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 3/4] net/af_xdp: make reserve/submit peek/release consistent

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> As David pointed out, if we reserve N slots for Tx, but only submit n
> slots, we would end up with an incorrect opinion of the number of available
> slots later, we also would get wrong idx when we call
> xsk_ring_prod__reserve next time. It also applies to
> xsk_ring_cons__peek()/xsk_ring_cons__release().
>
> This patch ensures that both reserve/submit and peek/release are
> consistent.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Suggested-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 79 +++++++++++++++--------------
>  1 file changed, 40 insertions(+), 39 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 8430921af..817092584 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
>  };
>
>  static inline int
> -reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
> +reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
>  {
>         struct xsk_ring_prod *fq = &umem->fq;
> +       void *addrs[reserve_size];
>         uint32_t idx;
> -       int i, ret;
> +       uint16_t i;
> +
> +       if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size,
> NULL)
> +                   != reserve_size) {
> +               AF_XDP_LOG(DEBUG, "Failed to get enough buffers for
> fq.\n");
> +               return -1;
> +       }
>
> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
> -       if (unlikely(!ret)) {
> -               AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
> -               return ret;
> +       if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
> +               AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs,
> +                               reserve_size, NULL);
> +               return -1;
>         }
>
>         for (i = 0; i < reserve_size; i++) {
>                 __u64 *fq_addr;
> -               void *addr = NULL;
> -               if (rte_ring_dequeue(umem->buf_ring, &addr)) {
> -                       i--;
> -                       break;
> -               }
> +
>                 fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
> -               *fq_addr = (uint64_t)addr;
> +               *fq_addr = (uint64_t)addrs[i];
>         }
>
> -       xsk_ring_prod__submit(fq, i);
> +       xsk_ring_prod__submit(fq, reserve_size);
>
>         return 0;
>  }
> @@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long dropped = 0;
>         unsigned long rx_bytes = 0;
> -       uint16_t count = 0;
>         int rcvd, i;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>
> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
> +               return 0;
> +
>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>         if (rcvd == 0)
> -               return 0;
> +               goto out;
>
>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>
> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
> 0))
> -               return 0;
> -
>         for (i = 0; i < rcvd; i++) {
>                 const struct xdp_desc *desc;
>                 uint64_t addr;
> @@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>                 rte_pktmbuf_pkt_len(mbufs[i]) = len;
>                 rte_pktmbuf_data_len(mbufs[i]) = len;
>                 rx_bytes += len;
> -               bufs[count++] = mbufs[i];
> +               bufs[i] = mbufs[i];
>
>                 rte_ring_enqueue(umem->buf_ring, (void *)addr);
>         }
> @@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         rxq->stats.rx_pkts += (rcvd - dropped);
>         rxq->stats.rx_bytes += rx_bytes;
>
> -       return count;
> +out:
> +       if (rcvd != nb_pkts)
> +               rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
> +                                    nb_pkts - rcvd);
> +
> +       return rcvd;
>  }
>
>  static void
> @@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbuf;
>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long tx_bytes = 0;
> -       int i, valid = 0;
> +       int i;
>         uint32_t idx_tx;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
> @@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         for (i = 0; i < nb_pkts; i++) {
>                 struct xdp_desc *desc;
>                 void *pkt;
> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
> -                                       - ETH_AF_XDP_DATA_HEADROOM;
> +
>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>                 mbuf = bufs[i];
> -               if (mbuf->pkt_len <= buf_len) {
> -                       desc->addr = (uint64_t)addrs[valid];
> -                       desc->len = mbuf->pkt_len;
> -                       pkt = xsk_umem__get_data(umem->mz->addr,
> -                                                desc->addr);
> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> -                              desc->len);
> -                       valid++;
> -                       tx_bytes += mbuf->pkt_len;
> -               }
> +
> +               desc->addr = (uint64_t)addrs[i];
> +               desc->len = mbuf->pkt_len;
> +               pkt = xsk_umem__get_data(umem->mz->addr,
> +                                        desc->addr);
> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> +                          desc->len);
> +               tx_bytes += mbuf->pkt_len;
> +
>                 rte_pktmbuf_free(mbuf);
>         }
>
> @@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         kick_tx(txq);
>
> -       if (valid < nb_pkts)
> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
> -                                nb_pkts - valid, NULL);
> -
> -       txq->stats.err_pkts += nb_pkts - valid;
> -       txq->stats.tx_pkts += valid;
> +       txq->stats.tx_pkts += nb_pkts;
>         txq->stats.tx_bytes += tx_bytes;
>
>         return nb_pkts;
> --
> 2.17.1
>
>

Reviewed-by: David Marchand <david.marchand@redhat.com>

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 3/4] net/af_xdp: make reserve/submit peek/release consistent

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> As David pointed out, if we reserve N slots for Tx, but only submit n
> slots, we would end up with an incorrect opinion of the number of available
> slots later, we also would get wrong idx when we call
> xsk_ring_prod__reserve next time. It also applies to
> xsk_ring_cons__peek()/xsk_ring_cons__release().
>
> This patch ensures that both reserve/submit and peek/release are
> consistent.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
> Suggested-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 79 +++++++++++++++--------------
>  1 file changed, 40 insertions(+), 39 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 8430921af..817092584 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
>  };
>
>  static inline int
> -reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
> +reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
>  {
>         struct xsk_ring_prod *fq = &umem->fq;
> +       void *addrs[reserve_size];
>         uint32_t idx;
> -       int i, ret;
> +       uint16_t i;
> +
> +       if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size,
> NULL)
> +                   != reserve_size) {
> +               AF_XDP_LOG(DEBUG, "Failed to get enough buffers for
> fq.\n");
> +               return -1;
> +       }
>
> -       ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
> -       if (unlikely(!ret)) {
> -               AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
> -               return ret;
> +       if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
> +               AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs,
> +                               reserve_size, NULL);
> +               return -1;
>         }
>
>         for (i = 0; i < reserve_size; i++) {
>                 __u64 *fq_addr;
> -               void *addr = NULL;
> -               if (rte_ring_dequeue(umem->buf_ring, &addr)) {
> -                       i--;
> -                       break;
> -               }
> +
>                 fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
> -               *fq_addr = (uint64_t)addr;
> +               *fq_addr = (uint64_t)addrs[i];
>         }
>
> -       xsk_ring_prod__submit(fq, i);
> +       xsk_ring_prod__submit(fq, reserve_size);
>
>         return 0;
>  }
> @@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long dropped = 0;
>         unsigned long rx_bytes = 0;
> -       uint16_t count = 0;
>         int rcvd, i;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
>
> +       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
> +               return 0;
> +
>         rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
>         if (rcvd == 0)
> -               return 0;
> +               goto out;
>
>         if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
>                 (void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
>
> -       if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) !=
> 0))
> -               return 0;
> -
>         for (i = 0; i < rcvd; i++) {
>                 const struct xdp_desc *desc;
>                 uint64_t addr;
> @@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>                 rte_pktmbuf_pkt_len(mbufs[i]) = len;
>                 rte_pktmbuf_data_len(mbufs[i]) = len;
>                 rx_bytes += len;
> -               bufs[count++] = mbufs[i];
> +               bufs[i] = mbufs[i];
>
>                 rte_ring_enqueue(umem->buf_ring, (void *)addr);
>         }
> @@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         rxq->stats.rx_pkts += (rcvd - dropped);
>         rxq->stats.rx_bytes += rx_bytes;
>
> -       return count;
> +out:
> +       if (rcvd != nb_pkts)
> +               rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
> +                                    nb_pkts - rcvd);
> +
> +       return rcvd;
>  }
>
>  static void
> @@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct rte_mbuf *mbuf;
>         void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
>         unsigned long tx_bytes = 0;
> -       int i, valid = 0;
> +       int i;
>         uint32_t idx_tx;
>
>         nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
> @@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         for (i = 0; i < nb_pkts; i++) {
>                 struct xdp_desc *desc;
>                 void *pkt;
> -               uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
> -                                       - ETH_AF_XDP_DATA_HEADROOM;
> +
>                 desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
>                 mbuf = bufs[i];
> -               if (mbuf->pkt_len <= buf_len) {
> -                       desc->addr = (uint64_t)addrs[valid];
> -                       desc->len = mbuf->pkt_len;
> -                       pkt = xsk_umem__get_data(umem->mz->addr,
> -                                                desc->addr);
> -                       rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> -                              desc->len);
> -                       valid++;
> -                       tx_bytes += mbuf->pkt_len;
> -               }
> +
> +               desc->addr = (uint64_t)addrs[i];
> +               desc->len = mbuf->pkt_len;
> +               pkt = xsk_umem__get_data(umem->mz->addr,
> +                                        desc->addr);
> +               rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
> +                          desc->len);
> +               tx_bytes += mbuf->pkt_len;
> +
>                 rte_pktmbuf_free(mbuf);
>         }
>
> @@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>
>         kick_tx(txq);
>
> -       if (valid < nb_pkts)
> -               rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
> -                                nb_pkts - valid, NULL);
> -
> -       txq->stats.err_pkts += nb_pkts - valid;
> -       txq->stats.tx_pkts += valid;
> +       txq->stats.tx_pkts += nb_pkts;
>         txq->stats.tx_bytes += tx_bytes;
>
>         return nb_pkts;
> --
> 2.17.1
>
>

Reviewed-by: David Marchand <david.marchand@redhat.com>

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 4/4] net/af_xdp: fix typos in Rx function

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> We should use ETH_AF_XDP_RX_BATCH_SIZE in Rx function rather than
> ETH_AF_XDP_TX_BATCH_SIZE.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
>
Reported-by: David Marchand <david.marchand@redhat.com>

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 4 ++--
>  1 file changed, 2 insertions(+), 2 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 817092584..ee10df0dc 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -175,12 +175,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct xsk_ring_prod *fq = &umem->fq;
>         uint32_t idx_rx = 0;
>         uint32_t free_thresh = fq->size >> 1;
> -       struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
> +       struct rte_mbuf *mbufs[ETH_AF_XDP_RX_BATCH_SIZE];
>         unsigned long dropped = 0;
>         unsigned long rx_bytes = 0;
>         int rcvd, i;
>
> -       nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
> +       nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_RX_BATCH_SIZE);
>
>         if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
>                 return 0;
> --
> 2.17.1
>

Reviewed-by: David Marchand <david.marchand@redhat.com>

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 4/4] net/af_xdp: fix typos in Rx function

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> We should use ETH_AF_XDP_RX_BATCH_SIZE in Rx function rather than
> ETH_AF_XDP_TX_BATCH_SIZE.
>
> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>
>
Reported-by: David Marchand <david.marchand@redhat.com>

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 4 ++--
>  1 file changed, 2 insertions(+), 2 deletions(-)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 817092584..ee10df0dc 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -175,12 +175,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs,
> uint16_t nb_pkts)
>         struct xsk_ring_prod *fq = &umem->fq;
>         uint32_t idx_rx = 0;
>         uint32_t free_thresh = fq->size >> 1;
> -       struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
> +       struct rte_mbuf *mbufs[ETH_AF_XDP_RX_BATCH_SIZE];
>         unsigned long dropped = 0;
>         unsigned long rx_bytes = 0;
>         int rcvd, i;
>
> -       nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
> +       nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_RX_BATCH_SIZE);
>
>         if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts)
> != 0))
>                 return 0;
> --
> 2.17.1
>

Reviewed-by: David Marchand <david.marchand@redhat.com>

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 2/4] net/af_xdp: specify minimal and maximal MTU

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> Since AF_XDP pmd doesn't support multi segment, we need to add a valid
> check in eth_dev_mtu_set.
>

How about:
Properly report mtu capability in port device info.


Reported-by: David Marchand <david.marchand@redhat.com>

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 12 ++++++++++++
>  1 file changed, 12 insertions(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..8430921af 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct
> rte_eth_dev_info *dev_info)
>         dev_info->max_rx_queues = 1;
>         dev_info->max_tx_queues = 1;
>
> +       dev_info->min_mtu = ETHER_MIN_MTU;
> +       dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE -
> ETH_AF_XDP_DATA_HEADROOM;
> +
>
        dev_info->default_rxportconf.nb_queues = 1;
>         dev_info->default_txportconf.nb_queues = 1;
>         dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
> @@ -654,6 +657,15 @@ eth_dev_mtu_set(struct rte_eth_dev *dev, uint16_t mtu)
>         int ret;
>         int s;
>
> +       if (mtu > ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM ||
> +                       mtu < ETHER_MIN_MTU) {
> +               AF_XDP_LOG(ERR, "Unsupported MTU of %d. "
> +                       "max mtu: %d, min mtu: %d", mtu,
> +                       ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM,
> +                       ETHER_MIN_MTU);
> +               return -EINVAL;
> +       }
> +
>

Sorry, I suppose my previous mail was confusing.
If you provide min/max_mtu, ethdev will enforce those checks for you and
you don't need to care about it.

See:
https://git.dpdk.org/dpdk/tree/lib/librte_ethdev/rte_ethdev.c?h=v19.05-rc1#n2630

With this block removed, you can add my review tag.

Thanks.

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 2/4] net/af_xdp: specify minimal and maximal MTU

[David Marchand]

On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> Since AF_XDP pmd doesn't support multi segment, we need to add a valid
> check in eth_dev_mtu_set.
>

How about:
Properly report mtu capability in port device info.


Reported-by: David Marchand <david.marchand@redhat.com>

Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
> ---
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 12 ++++++++++++
>  1 file changed, 12 insertions(+)
>
> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
> b/drivers/net/af_xdp/rte_eth_af_xdp.c
> index 5cc643ce2..8430921af 100644
> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
> @@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct
> rte_eth_dev_info *dev_info)
>         dev_info->max_rx_queues = 1;
>         dev_info->max_tx_queues = 1;
>
> +       dev_info->min_mtu = ETHER_MIN_MTU;
> +       dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE -
> ETH_AF_XDP_DATA_HEADROOM;
> +
>
        dev_info->default_rxportconf.nb_queues = 1;
>         dev_info->default_txportconf.nb_queues = 1;
>         dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
> @@ -654,6 +657,15 @@ eth_dev_mtu_set(struct rte_eth_dev *dev, uint16_t mtu)
>         int ret;
>         int s;
>
> +       if (mtu > ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM ||
> +                       mtu < ETHER_MIN_MTU) {
> +               AF_XDP_LOG(ERR, "Unsupported MTU of %d. "
> +                       "max mtu: %d, min mtu: %d", mtu,
> +                       ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM,
> +                       ETHER_MIN_MTU);
> +               return -EINVAL;
> +       }
> +
>

Sorry, I suppose my previous mail was confusing.
If you provide min/max_mtu, ethdev will enforce those checks for you and
you don't need to care about it.

See:
https://git.dpdk.org/dpdk/tree/lib/librte_ethdev/rte_ethdev.c?h=v19.05-rc1#n2630

With this block removed, you can add my review tag.

Thanks.

-- 
David Marchand

Re: [dpdk-dev] [PATCH v3 2/4] net/af_xdp: specify minimal and maximal MTU

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> Since AF_XDP pmd doesn't support multi segment, we need to add a valid
>> check in eth_dev_mtu_set.
>>
>
>How about:
>Properly report mtu capability in port device info.

Sounds better.

>
>
>Reported-by: David Marchand <david.marchand@redhat.com>
>
>Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 12 ++++++++++++
>>  1 file changed, 12 insertions(+)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..8430921af 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct
>> rte_eth_dev_info *dev_info)
>>         dev_info->max_rx_queues = 1;
>>         dev_info->max_tx_queues = 1;
>>
>> +       dev_info->min_mtu = ETHER_MIN_MTU;
>> +       dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE -
>> ETH_AF_XDP_DATA_HEADROOM;
>> +
>>
>        dev_info->default_rxportconf.nb_queues = 1;
>>         dev_info->default_txportconf.nb_queues = 1;
>>         dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
>> @@ -654,6 +657,15 @@ eth_dev_mtu_set(struct rte_eth_dev *dev, uint16_t mtu)
>>         int ret;
>>         int s;
>>
>> +       if (mtu > ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM ||
>> +                       mtu < ETHER_MIN_MTU) {
>> +               AF_XDP_LOG(ERR, "Unsupported MTU of %d. "
>> +                       "max mtu: %d, min mtu: %d", mtu,
>> +                       ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM,
>> +                       ETHER_MIN_MTU);
>> +               return -EINVAL;
>> +       }
>> +
>>
>
>Sorry, I suppose my previous mail was confusing.
>If you provide min/max_mtu, ethdev will enforce those checks for you and
>you don't need to care about it.
>
>See:
>https://git.dpdk.org/dpdk/tree/lib/librte_ethdev/rte_ethdev.c?h=v19.05-rc1#n2630
>
>With this block removed, you can add my review tag.

Got it, will remove it in next version.

Thanks,
Xiaolong
>
>Thanks.
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH v3 2/4] net/af_xdp: specify minimal and maximal MTU

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> Since AF_XDP pmd doesn't support multi segment, we need to add a valid
>> check in eth_dev_mtu_set.
>>
>
>How about:
>Properly report mtu capability in port device info.

Sounds better.

>
>
>Reported-by: David Marchand <david.marchand@redhat.com>
>
>Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 12 ++++++++++++
>>  1 file changed, 12 insertions(+)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 5cc643ce2..8430921af 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct
>> rte_eth_dev_info *dev_info)
>>         dev_info->max_rx_queues = 1;
>>         dev_info->max_tx_queues = 1;
>>
>> +       dev_info->min_mtu = ETHER_MIN_MTU;
>> +       dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE -
>> ETH_AF_XDP_DATA_HEADROOM;
>> +
>>
>        dev_info->default_rxportconf.nb_queues = 1;
>>         dev_info->default_txportconf.nb_queues = 1;
>>         dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
>> @@ -654,6 +657,15 @@ eth_dev_mtu_set(struct rte_eth_dev *dev, uint16_t mtu)
>>         int ret;
>>         int s;
>>
>> +       if (mtu > ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM ||
>> +                       mtu < ETHER_MIN_MTU) {
>> +               AF_XDP_LOG(ERR, "Unsupported MTU of %d. "
>> +                       "max mtu: %d, min mtu: %d", mtu,
>> +                       ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM,
>> +                       ETHER_MIN_MTU);
>> +               return -EINVAL;
>> +       }
>> +
>>
>
>Sorry, I suppose my previous mail was confusing.
>If you provide min/max_mtu, ethdev will enforce those checks for you and
>you don't need to care about it.
>
>See:
>https://git.dpdk.org/dpdk/tree/lib/librte_ethdev/rte_ethdev.c?h=v19.05-rc1#n2630
>
>With this block removed, you can add my review tag.

Got it, will remove it in next version.

Thanks,
Xiaolong
>
>Thanks.
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH v3 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> When it fails to allocate enough slots in Tx queue for transmitting
>> packets, we need to return the dequeued addrs to buf ring.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>>  1 file changed, 1 insertion(+)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 007a1c6b4..5cc643ce2 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
>> {
>>                 kick_tx(txq);
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
>> NULL);
>>                 return 0;
>>         }
>>
>> --
>> 2.17.1
>>
>
>When there is no change, please keep Review tags from previous versions.
>
>Reviewed-by: David Marchand <david.marchand@redhat.com>

Got it, will add your review tag in next version.

Thanks,
Xiaolong
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH v3 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Wed, Apr 17, 2019 at 11:02 AM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> When it fails to allocate enough slots in Tx queue for transmitting
>> packets, we need to return the dequeued addrs to buf ring.
>>
>> Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")
>>
>> Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
>> ---
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
>>  1 file changed, 1 insertion(+)
>>
>> diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> index 007a1c6b4..5cc643ce2 100644
>> --- a/drivers/net/af_xdp/rte_eth_af_xdp.c
>> +++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
>> @@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs,
>> uint16_t nb_pkts)
>>
>>         if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts)
>> {
>>                 kick_tx(txq);
>> +               rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts,
>> NULL);
>>                 return 0;
>>         }
>>
>> --
>> 2.17.1
>>
>
>When there is no change, please keep Review tags from previous versions.
>
>Reviewed-by: David Marchand <david.marchand@redhat.com>

Got it, will add your review tag in next version.

Thanks,
Xiaolong
>
>-- 
>David Marchand

[dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[Xiaolong Ye]

This patchset provides some fixes to af_xdp pmd, at first, I just added
a simple error handling when Tx queue allocation fails, then David
suggested a better way to do it and pointed out the inconsistent issue
of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
queue), the third patch addressed this.

v4 changes:

- remove unnecessary mtu valid check in eth_dev_mtu_set
- add Reported-by and Reviewed-by tags of David

v3 changes:

- address David's review comments
- add one patch to specify the mtu range
- add one fix patch for typo

v2 changes:

- adopt David's suggestion to refactor the code

Xiaolong Ye (4):
  net/af_xdp: enqueue buf ring when allocate Tx queue fails
  net/af_xdp: specify minimal and maximal MTU
  net/af_xdp: make reserve/submit peek/release consistent
  net/af_xdp: fix typos in Rx function

 drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
 1 file changed, 46 insertions(+), 41 deletions(-)

-- 
2.17.1

[dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[Xiaolong Ye]

This patchset provides some fixes to af_xdp pmd, at first, I just added
a simple error handling when Tx queue allocation fails, then David
suggested a better way to do it and pointed out the inconsistent issue
of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
queue), the third patch addressed this.

v4 changes:

- remove unnecessary mtu valid check in eth_dev_mtu_set
- add Reported-by and Reviewed-by tags of David

v3 changes:

- address David's review comments
- add one patch to specify the mtu range
- add one fix patch for typo

v2 changes:

- adopt David's suggestion to refactor the code

Xiaolong Ye (4):
  net/af_xdp: enqueue buf ring when allocate Tx queue fails
  net/af_xdp: specify minimal and maximal MTU
  net/af_xdp: make reserve/submit peek/release consistent
  net/af_xdp: fix typos in Rx function

 drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
 1 file changed, 46 insertions(+), 41 deletions(-)

-- 
2.17.1

[dpdk-dev] [PATCH v4 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH v4 1/4] net/af_xdp: enqueue buf ring when allocate Tx queue fails

[Xiaolong Ye]

When it fails to allocate enough slots in Tx queue for transmitting
packets, we need to return the dequeued addrs to buf ring.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 007a1c6b4..5cc643ce2 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -276,6 +276,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	if (xsk_ring_prod__reserve(&txq->tx, nb_pkts, &idx_tx) != nb_pkts) {
 		kick_tx(txq);
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs, nb_pkts, NULL);
 		return 0;
 	}
 
-- 
2.17.1

[dpdk-dev] [PATCH v4 2/4] net/af_xdp: specify minimal and maximal MTU

[Xiaolong Ye]

Properly report mtu capability in port device info.

Reported-by: David Marchand <david.marchand@redhat.com>
Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..c46916bbe 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct rte_eth_dev_info *dev_info)
 	dev_info->max_rx_queues = 1;
 	dev_info->max_tx_queues = 1;
 
+	dev_info->min_mtu = ETHER_MIN_MTU;
+	dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM;
+
 	dev_info->default_rxportconf.nb_queues = 1;
 	dev_info->default_txportconf.nb_queues = 1;
 	dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
-- 
2.17.1

[dpdk-dev] [PATCH v4 2/4] net/af_xdp: specify minimal and maximal MTU

[Xiaolong Ye]

Properly report mtu capability in port device info.

Reported-by: David Marchand <david.marchand@redhat.com>
Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 5cc643ce2..c46916bbe 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -351,6 +351,9 @@ eth_dev_info(struct rte_eth_dev *dev, struct rte_eth_dev_info *dev_info)
 	dev_info->max_rx_queues = 1;
 	dev_info->max_tx_queues = 1;
 
+	dev_info->min_mtu = ETHER_MIN_MTU;
+	dev_info->max_mtu = ETH_AF_XDP_FRAME_SIZE - ETH_AF_XDP_DATA_HEADROOM;
+
 	dev_info->default_rxportconf.nb_queues = 1;
 	dev_info->default_txportconf.nb_queues = 1;
 	dev_info->default_rxportconf.ring_size = ETH_AF_XDP_DFLT_NUM_DESCS;
-- 
2.17.1

[dpdk-dev] [PATCH v4 3/4] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots for Tx, but only submit n
slots, we would end up with an incorrect opinion of the number of available
slots later, we also would get wrong idx when we call
xsk_ring_prod__reserve next time. It also applies to
xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Suggested-by: David Marchand <david.marchand@redhat.com>
Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 79 +++++++++++++++--------------
 1 file changed, 40 insertions(+), 39 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index c46916bbe..6a0096523 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
 };
 
 static inline int
-reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
+reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
+	void *addrs[reserve_size];
 	uint32_t idx;
-	int i, ret;
+	uint16_t i;
+
+	if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size, NULL)
+		    != reserve_size) {
+		AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
+		return -1;
+	}
 
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
-		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
+	if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
+		AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs,
+				reserve_size, NULL);
+		return -1;
 	}
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
-		void *addr = NULL;
-		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
-			break;
-		}
+
 		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
-		*fq_addr = (uint64_t)addr;
+		*fq_addr = (uint64_t)addrs[i];
 	}
 
-	xsk_ring_prod__submit(fq, i);
+	xsk_ring_prod__submit(fq, reserve_size);
 
 	return 0;
 }
@@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
-	uint16_t count = 0;
 	int rcvd, i;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
-		return 0;
+		goto out;
 
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_pkt_len(mbufs[i]) = len;
 		rte_pktmbuf_data_len(mbufs[i]) = len;
 		rx_bytes += len;
-		bufs[count++] = mbufs[i];
+		bufs[i] = mbufs[i];
 
 		rte_ring_enqueue(umem->buf_ring, (void *)addr);
 	}
@@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
 
-	return count;
+out:
+	if (rcvd != nb_pkts)
+		rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
+				     nb_pkts - rcvd);
+
+	return rcvd;
 }
 
 static void
@@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
 	uint32_t idx_tx;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
@@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	for (i = 0; i < nb_pkts; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
+
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
 		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
+
 		rte_pktmbuf_free(mbuf);
 	}
 
@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.tx_pkts += nb_pkts;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

[dpdk-dev] [PATCH v4 3/4] net/af_xdp: make reserve/submit peek/release consistent

[Xiaolong Ye]

As David pointed out, if we reserve N slots for Tx, but only submit n
slots, we would end up with an incorrect opinion of the number of available
slots later, we also would get wrong idx when we call
xsk_ring_prod__reserve next time. It also applies to
xsk_ring_cons__peek()/xsk_ring_cons__release().

This patch ensures that both reserve/submit and peek/release are
consistent.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Suggested-by: David Marchand <david.marchand@redhat.com>
Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 79 +++++++++++++++--------------
 1 file changed, 40 insertions(+), 39 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index c46916bbe..6a0096523 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -134,30 +134,34 @@ static const struct rte_eth_link pmd_link = {
 };
 
 static inline int
-reserve_fill_queue(struct xsk_umem_info *umem, int reserve_size)
+reserve_fill_queue(struct xsk_umem_info *umem, uint16_t reserve_size)
 {
 	struct xsk_ring_prod *fq = &umem->fq;
+	void *addrs[reserve_size];
 	uint32_t idx;
-	int i, ret;
+	uint16_t i;
+
+	if (rte_ring_dequeue_bulk(umem->buf_ring, addrs, reserve_size, NULL)
+		    != reserve_size) {
+		AF_XDP_LOG(DEBUG, "Failed to get enough buffers for fq.\n");
+		return -1;
+	}
 
-	ret = xsk_ring_prod__reserve(fq, reserve_size, &idx);
-	if (unlikely(!ret)) {
-		AF_XDP_LOG(ERR, "Failed to reserve enough fq descs.\n");
-		return ret;
+	if (unlikely(!xsk_ring_prod__reserve(fq, reserve_size, &idx))) {
+		AF_XDP_LOG(DEBUG, "Failed to reserve enough fq descs.\n");
+		rte_ring_enqueue_bulk(umem->buf_ring, addrs,
+				reserve_size, NULL);
+		return -1;
 	}
 
 	for (i = 0; i < reserve_size; i++) {
 		__u64 *fq_addr;
-		void *addr = NULL;
-		if (rte_ring_dequeue(umem->buf_ring, &addr)) {
-			i--;
-			break;
-		}
+
 		fq_addr = xsk_ring_prod__fill_addr(fq, idx++);
-		*fq_addr = (uint64_t)addr;
+		*fq_addr = (uint64_t)addrs[i];
 	}
 
-	xsk_ring_prod__submit(fq, i);
+	xsk_ring_prod__submit(fq, reserve_size);
 
 	return 0;
 }
@@ -174,21 +178,20 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
-	uint16_t count = 0;
 	int rcvd, i;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
 
+	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
+		return 0;
+
 	rcvd = xsk_ring_cons__peek(rx, nb_pkts, &idx_rx);
 	if (rcvd == 0)
-		return 0;
+		goto out;
 
 	if (xsk_prod_nb_free(fq, free_thresh) >= free_thresh)
 		(void)reserve_fill_queue(umem, ETH_AF_XDP_RX_BATCH_SIZE);
 
-	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, rcvd) != 0))
-		return 0;
-
 	for (i = 0; i < rcvd; i++) {
 		const struct xdp_desc *desc;
 		uint64_t addr;
@@ -204,7 +207,7 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 		rte_pktmbuf_pkt_len(mbufs[i]) = len;
 		rte_pktmbuf_data_len(mbufs[i]) = len;
 		rx_bytes += len;
-		bufs[count++] = mbufs[i];
+		bufs[i] = mbufs[i];
 
 		rte_ring_enqueue(umem->buf_ring, (void *)addr);
 	}
@@ -215,7 +218,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	rxq->stats.rx_pkts += (rcvd - dropped);
 	rxq->stats.rx_bytes += rx_bytes;
 
-	return count;
+out:
+	if (rcvd != nb_pkts)
+		rte_mempool_put_bulk(rxq->mb_pool, (void **)&mbufs[rcvd],
+				     nb_pkts - rcvd);
+
+	return rcvd;
 }
 
 static void
@@ -262,7 +270,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct rte_mbuf *mbuf;
 	void *addrs[ETH_AF_XDP_TX_BATCH_SIZE];
 	unsigned long tx_bytes = 0;
-	int i, valid = 0;
+	int i;
 	uint32_t idx_tx;
 
 	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
@@ -283,20 +291,18 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	for (i = 0; i < nb_pkts; i++) {
 		struct xdp_desc *desc;
 		void *pkt;
-		uint32_t buf_len = ETH_AF_XDP_FRAME_SIZE
-					- ETH_AF_XDP_DATA_HEADROOM;
+
 		desc = xsk_ring_prod__tx_desc(&txq->tx, idx_tx + i);
 		mbuf = bufs[i];
-		if (mbuf->pkt_len <= buf_len) {
-			desc->addr = (uint64_t)addrs[valid];
-			desc->len = mbuf->pkt_len;
-			pkt = xsk_umem__get_data(umem->mz->addr,
-						 desc->addr);
-			rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
-			       desc->len);
-			valid++;
-			tx_bytes += mbuf->pkt_len;
-		}
+
+		desc->addr = (uint64_t)addrs[i];
+		desc->len = mbuf->pkt_len;
+		pkt = xsk_umem__get_data(umem->mz->addr,
+					 desc->addr);
+		rte_memcpy(pkt, rte_pktmbuf_mtod(mbuf, void *),
+			   desc->len);
+		tx_bytes += mbuf->pkt_len;
+
 		rte_pktmbuf_free(mbuf);
 	}
 
@@ -304,12 +310,7 @@ eth_af_xdp_tx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	kick_tx(txq);
 
-	if (valid < nb_pkts)
-		rte_ring_enqueue_bulk(umem->buf_ring, &addrs[valid],
-				 nb_pkts - valid, NULL);
-
-	txq->stats.err_pkts += nb_pkts - valid;
-	txq->stats.tx_pkts += valid;
+	txq->stats.tx_pkts += nb_pkts;
 	txq->stats.tx_bytes += tx_bytes;
 
 	return nb_pkts;
-- 
2.17.1

[dpdk-dev] [PATCH v4 4/4] net/af_xdp: fix typos in Rx function

[Xiaolong Ye]

We should use ETH_AF_XDP_RX_BATCH_SIZE in Rx function rather than
ETH_AF_XDP_TX_BATCH_SIZE.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reported-by: David Marchand <david.marchand@redhat.com>
Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 6a0096523..497e2cfde 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -175,12 +175,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct xsk_ring_prod *fq = &umem->fq;
 	uint32_t idx_rx = 0;
 	uint32_t free_thresh = fq->size >> 1;
-	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
+	struct rte_mbuf *mbufs[ETH_AF_XDP_RX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
 	int rcvd, i;
 
-	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
+	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_RX_BATCH_SIZE);
 
 	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
 		return 0;
-- 
2.17.1

[dpdk-dev] [PATCH v4 4/4] net/af_xdp: fix typos in Rx function

[Xiaolong Ye]

We should use ETH_AF_XDP_RX_BATCH_SIZE in Rx function rather than
ETH_AF_XDP_TX_BATCH_SIZE.

Fixes: f1debd77efaf ("net/af_xdp: introduce AF_XDP PMD")

Reported-by: David Marchand <david.marchand@redhat.com>
Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Xiaolong Ye <xiaolong.ye@intel.com>
---
 drivers/net/af_xdp/rte_eth_af_xdp.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/drivers/net/af_xdp/rte_eth_af_xdp.c b/drivers/net/af_xdp/rte_eth_af_xdp.c
index 6a0096523..497e2cfde 100644
--- a/drivers/net/af_xdp/rte_eth_af_xdp.c
+++ b/drivers/net/af_xdp/rte_eth_af_xdp.c
@@ -175,12 +175,12 @@ eth_af_xdp_rx(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct xsk_ring_prod *fq = &umem->fq;
 	uint32_t idx_rx = 0;
 	uint32_t free_thresh = fq->size >> 1;
-	struct rte_mbuf *mbufs[ETH_AF_XDP_TX_BATCH_SIZE];
+	struct rte_mbuf *mbufs[ETH_AF_XDP_RX_BATCH_SIZE];
 	unsigned long dropped = 0;
 	unsigned long rx_bytes = 0;
 	int rcvd, i;
 
-	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_TX_BATCH_SIZE);
+	nb_pkts = RTE_MIN(nb_pkts, ETH_AF_XDP_RX_BATCH_SIZE);
 
 	if (unlikely(rte_pktmbuf_alloc_bulk(rxq->mb_pool, mbufs, nb_pkts) != 0))
 		return 0;
-- 
2.17.1

Re: [dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[David Marchand]

On Wed, Apr 17, 2019 at 3:55 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> This patchset provides some fixes to af_xdp pmd, at first, I just added
> a simple error handling when Tx queue allocation fails, then David
> suggested a better way to do it and pointed out the inconsistent issue
> of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
> queue), the third patch addressed this.
>
> v4 changes:
>
> - remove unnecessary mtu valid check in eth_dev_mtu_set
> - add Reported-by and Reviewed-by tags of David
>
> v3 changes:
>
> - address David's review comments
> - add one patch to specify the mtu range
> - add one fix patch for typo
>
> v2 changes:
>
> - adopt David's suggestion to refactor the code
>
> Xiaolong Ye (4):
>   net/af_xdp: enqueue buf ring when allocate Tx queue fails
>   net/af_xdp: specify minimal and maximal MTU
>   net/af_xdp: make reserve/submit peek/release consistent
>   net/af_xdp: fix typos in Rx function
>
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
>  1 file changed, 46 insertions(+), 41 deletions(-)
>
> --
> 2.17.1
>
>
LGTM.
Thanks Xiaolong.


-- 
David Marchand

Re: [dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[David Marchand]

On Wed, Apr 17, 2019 at 3:55 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:

> This patchset provides some fixes to af_xdp pmd, at first, I just added
> a simple error handling when Tx queue allocation fails, then David
> suggested a better way to do it and pointed out the inconsistent issue
> of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
> queue), the third patch addressed this.
>
> v4 changes:
>
> - remove unnecessary mtu valid check in eth_dev_mtu_set
> - add Reported-by and Reviewed-by tags of David
>
> v3 changes:
>
> - address David's review comments
> - add one patch to specify the mtu range
> - add one fix patch for typo
>
> v2 changes:
>
> - adopt David's suggestion to refactor the code
>
> Xiaolong Ye (4):
>   net/af_xdp: enqueue buf ring when allocate Tx queue fails
>   net/af_xdp: specify minimal and maximal MTU
>   net/af_xdp: make reserve/submit peek/release consistent
>   net/af_xdp: fix typos in Rx function
>
>  drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
>  1 file changed, 46 insertions(+), 41 deletions(-)
>
> --
> 2.17.1
>
>
LGTM.
Thanks Xiaolong.


-- 
David Marchand

Re: [dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Wed, Apr 17, 2019 at 3:55 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> This patchset provides some fixes to af_xdp pmd, at first, I just added
>> a simple error handling when Tx queue allocation fails, then David
>> suggested a better way to do it and pointed out the inconsistent issue
>> of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
>> queue), the third patch addressed this.
>>
>> v4 changes:
>>
>> - remove unnecessary mtu valid check in eth_dev_mtu_set
>> - add Reported-by and Reviewed-by tags of David
>>
>> v3 changes:
>>
>> - address David's review comments
>> - add one patch to specify the mtu range
>> - add one fix patch for typo
>>
>> v2 changes:
>>
>> - adopt David's suggestion to refactor the code
>>
>> Xiaolong Ye (4):
>>   net/af_xdp: enqueue buf ring when allocate Tx queue fails
>>   net/af_xdp: specify minimal and maximal MTU
>>   net/af_xdp: make reserve/submit peek/release consistent
>>   net/af_xdp: fix typos in Rx function
>>
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
>>  1 file changed, 46 insertions(+), 41 deletions(-)
>>
>> --
>> 2.17.1
>>
>>
>LGTM.
>Thanks Xiaolong.

Thanks a lot for your guidance. :)

>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[Ye Xiaolong]

On 04/17, David Marchand wrote:
>On Wed, Apr 17, 2019 at 3:55 PM Xiaolong Ye <xiaolong.ye@intel.com> wrote:
>
>> This patchset provides some fixes to af_xdp pmd, at first, I just added
>> a simple error handling when Tx queue allocation fails, then David
>> suggested a better way to do it and pointed out the inconsistent issue
>> of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
>> queue), the third patch addressed this.
>>
>> v4 changes:
>>
>> - remove unnecessary mtu valid check in eth_dev_mtu_set
>> - add Reported-by and Reviewed-by tags of David
>>
>> v3 changes:
>>
>> - address David's review comments
>> - add one patch to specify the mtu range
>> - add one fix patch for typo
>>
>> v2 changes:
>>
>> - adopt David's suggestion to refactor the code
>>
>> Xiaolong Ye (4):
>>   net/af_xdp: enqueue buf ring when allocate Tx queue fails
>>   net/af_xdp: specify minimal and maximal MTU
>>   net/af_xdp: make reserve/submit peek/release consistent
>>   net/af_xdp: fix typos in Rx function
>>
>>  drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
>>  1 file changed, 46 insertions(+), 41 deletions(-)
>>
>> --
>> 2.17.1
>>
>>
>LGTM.
>Thanks Xiaolong.

Thanks a lot for your guidance. :)

>
>
>-- 
>David Marchand

Re: [dpdk-dev] [PATCH v4 4/4] net/af_xdp: fix typos in Rx function

[Rami Rosen]

Acked-by: Rami Rosen <ramirose@gmail.com>

Re: [dpdk-dev] [PATCH v4 4/4] net/af_xdp: fix typos in Rx function

[Rami Rosen]

Acked-by: Rami Rosen <ramirose@gmail.com>

Re: [dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[Ferruh Yigit]

On 4/17/2019 3:02 PM, David Marchand wrote:
> On Wed, Apr 17, 2019 at 3:55 PM Xiaolong Ye <xiaolong.ye@intel.com
> <mailto:xiaolong.ye@intel.com>> wrote:
> 
>     This patchset provides some fixes to af_xdp pmd, at first, I just added
>     a simple error handling when Tx queue allocation fails, then David
>     suggested a better way to do it and pointed out the inconsistent issue
>     of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
>     queue), the third patch addressed this.
> 
>     v4 changes:
> 
>     - remove unnecessary mtu valid check in eth_dev_mtu_set
>     - add Reported-by and Reviewed-by tags of David
> 
>     v3 changes:
> 
>     - address David's review comments
>     - add one patch to specify the mtu range
>     - add one fix patch for typo
> 
>     v2 changes:
> 
>     - adopt David's suggestion to refactor the code
> 
>     Xiaolong Ye (4):
>       net/af_xdp: enqueue buf ring when allocate Tx queue fails
>       net/af_xdp: specify minimal and maximal MTU
>       net/af_xdp: make reserve/submit peek/release consistent
>       net/af_xdp: fix typos in Rx function
> 
>      drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
>      1 file changed, 46 insertions(+), 41 deletions(-)
> 
>     -- 
>     2.17.1
> 
> 
> LGTM.
> Thanks Xiaolong.

Series applied to dpdk-next-net/master, thanks.

Re: [dpdk-dev] [PATCH v4 0/4] some fixes for AF_XDP pmd

[Ferruh Yigit]

On 4/17/2019 3:02 PM, David Marchand wrote:
> On Wed, Apr 17, 2019 at 3:55 PM Xiaolong Ye <xiaolong.ye@intel.com
> <mailto:xiaolong.ye@intel.com>> wrote:
> 
>     This patchset provides some fixes to af_xdp pmd, at first, I just added
>     a simple error handling when Tx queue allocation fails, then David
>     suggested a better way to do it and pointed out the inconsistent issue
>     of reserve/submit ops (for Tx queue) and peek/release ops (for Rx
>     queue), the third patch addressed this.
> 
>     v4 changes:
> 
>     - remove unnecessary mtu valid check in eth_dev_mtu_set
>     - add Reported-by and Reviewed-by tags of David
> 
>     v3 changes:
> 
>     - address David's review comments
>     - add one patch to specify the mtu range
>     - add one fix patch for typo
> 
>     v2 changes:
> 
>     - adopt David's suggestion to refactor the code
> 
>     Xiaolong Ye (4):
>       net/af_xdp: enqueue buf ring when allocate Tx queue fails
>       net/af_xdp: specify minimal and maximal MTU
>       net/af_xdp: make reserve/submit peek/release consistent
>       net/af_xdp: fix typos in Rx function
> 
>      drivers/net/af_xdp/rte_eth_af_xdp.c | 87 +++++++++++++++--------------
>      1 file changed, 46 insertions(+), 41 deletions(-)
> 
>     -- 
>     2.17.1
> 
> 
> LGTM.
> Thanks Xiaolong.

Series applied to dpdk-next-net/master, thanks.