Re: [PATCH bpf v2] xdp: fix race on generic receive path

[Magnus Karlsson]

On Wed, Jul 3, 2019 at 2:09 PM Ilya Maximets <i.maximets@xxxxxxxxxxx> wrote:
>
> Unlike driver mode, generic xdp receive could be triggered
> by different threads on different CPU cores at the same time
> leading to the fill and rx queue breakage. For example, this
> could happen while sending packets from two processes to the
> first interface of veth pair while the second part of it is
> open with AF_XDP socket.
>
> Need to take a lock for each generic receive to avoid race.

I measured the performance degradation of rxdrop on my local machine
and it went from 2.19 to 2.08, so roughly a 5% drop. I think we can
live with this in XDP_SKB mode. If we at some later point in time need
to boost performance in this mode, let us look at it then from a
broader perspective and find the most low hanging fruit.

Thanks Ilya for this fix.

Acked-by: Magnus Karlsson <magnus.karlsson@xxxxxxxxx>

> Fixes: c497176cb2e4 ("xsk: add Rx receive functions and poll support")
> Signed-off-by: Ilya Maximets <i.maximets@xxxxxxxxxxx>
> ---
>
> Version 2:
>     * spin_lock_irqsave --> spin_lock_bh.
>
>  include/net/xdp_sock.h |  2 ++
>  net/xdp/xsk.c          | 31 ++++++++++++++++++++++---------
>  2 files changed, 24 insertions(+), 9 deletions(-)
>
> diff --git a/include/net/xdp_sock.h b/include/net/xdp_sock.h
> index d074b6d60f8a..ac3c047d058c 100644
> --- a/include/net/xdp_sock.h
> +++ b/include/net/xdp_sock.h
> @@ -67,6 +67,8 @@ struct xdp_sock {
>          * in the SKB destructor callback.
>          */
>         spinlock_t tx_completion_lock;
> +       /* Protects generic receive. */
> +       spinlock_t rx_lock;
>         u64 rx_dropped;
>  };
>
> diff --git a/net/xdp/xsk.c b/net/xdp/xsk.c
> index a14e8864e4fa..5e0637db92ea 100644
> --- a/net/xdp/xsk.c
> +++ b/net/xdp/xsk.c
> @@ -123,13 +123,17 @@ int xsk_generic_rcv(struct xdp_sock *xs, struct xdp_buff *xdp)
>         u64 addr;
>         int err;
>
> -       if (xs->dev != xdp->rxq->dev || xs->queue_id != xdp->rxq->queue_index)
> -               return -EINVAL;
> +       spin_lock_bh(&xs->rx_lock);
> +
> +       if (xs->dev != xdp->rxq->dev || xs->queue_id != xdp->rxq->queue_index) {
> +               err = -EINVAL;
> +               goto out_unlock;
> +       }
>
>         if (!xskq_peek_addr(xs->umem->fq, &addr) ||
>             len > xs->umem->chunk_size_nohr - XDP_PACKET_HEADROOM) {
> -               xs->rx_dropped++;
> -               return -ENOSPC;
> +               err = -ENOSPC;
> +               goto out_drop;
>         }
>
>         addr += xs->umem->headroom;
> @@ -138,13 +142,21 @@ int xsk_generic_rcv(struct xdp_sock *xs, struct xdp_buff *xdp)
>         memcpy(buffer, xdp->data_meta, len + metalen);
>         addr += metalen;
>         err = xskq_produce_batch_desc(xs->rx, addr, len);
> -       if (!err) {
> -               xskq_discard_addr(xs->umem->fq);
> -               xsk_flush(xs);
> -               return 0;
> -       }
> +       if (err)
> +               goto out_drop;
> +
> +       xskq_discard_addr(xs->umem->fq);
> +       xskq_produce_flush_desc(xs->rx);
>
> +       spin_unlock_bh(&xs->rx_lock);
> +
> +       xs->sk.sk_data_ready(&xs->sk);
> +       return 0;
> +
> +out_drop:
>         xs->rx_dropped++;
> +out_unlock:
> +       spin_unlock_bh(&xs->rx_lock);
>         return err;
>  }
>
> @@ -765,6 +777,7 @@ static int xsk_create(struct net *net, struct socket *sock, int protocol,
>
>         xs = xdp_sk(sk);
>         mutex_init(&xs->mutex);
> +       spin_lock_init(&xs->rx_lock);
>         spin_lock_init(&xs->tx_completion_lock);
>
>         mutex_lock(&net->xdp.lock);
> --
> 2.17.1
>