RE: [RFC 01/32] bpf: replace deprecated SMP barriers with C11 fences

[Konstantin Ananyev <konstantin.ananyev@huawei.com>]

> 
> This commit replaces the deprecated DPDK SMP memory barriers in the
> BPF packet processing module with standard C11 atomic operations.
> 
> Rather than relying on standalone explicit fences (which can introduce
> unnecessary overhead and are less idiomatic in C11), this patch
> embeds the memory ordering directly into the atomic operations:
> 
> 1. The usage counter (`cbi->use`) now utilizes `rte_atomic_fetch_add_explicit`.
>    Entering the callback uses `acquire` semantics, and exiting uses
>    `release` semantics, forming a proper lightweight lock without standalone
> fences.

While atomic_add() is possible here, it is really unnecessary, as whole
DPDK ethdev RX/TX path assumes that only one thread at a time can do RX/TX
on given HW queue. 
So only one data-path can update cbi->use.
Memory fence (or it's analog) is necessary here to make sure that it would happen
before any reads to other fields would happen.
I.E. we don't need a proper lock semantics here, we just need to make sure that
STORE/LOAD re-ordering will not happen. 

> 2. The callback pointer (`cbi->cb`) is marked as `RTE_ATOMIC` to prevent
>    data races between the datapath and control path. It is updated
>    and read using explicit atomic loads/stores with `acquire`/`release` semantics.

That is totally unnecessary and even contradicts with #1 above:
- we don't need a full lock protection here for cbi 
- in case we really do, then atomic load wouldn't help us here as for such case
  we would need a proper mutex-like lock to protect all fields in struct bpf_eth_cbi.
  But as I said, we don't.
   
So in summary I am against those changes.

Just a bit of history here: what is implemented here is sort-of hand-written
analog of RCU QSBR mechanism. It was done before RCU lib was introduced
into the project.
Probably the right ways here would be to make ethdev RX/TX callback
API thread-safe via using our own lib/rcu. Then all this code around cbi->use
can be simply removed. But that's probably a subject for separate patch series. 

> ---
>  lib/bpf/bpf_pkt.c | 61 +++++++++++++++++++++++++++--------------------
>  1 file changed, 35 insertions(+), 26 deletions(-)
> 
> diff --git a/lib/bpf/bpf_pkt.c b/lib/bpf/bpf_pkt.c
> index f072fdaaed..a831b5ad86 100644
> --- a/lib/bpf/bpf_pkt.c
> +++ b/lib/bpf/bpf_pkt.c
> @@ -30,7 +30,7 @@
>  struct __rte_cache_aligned bpf_eth_cbi {
>  	/* used by both data & control path */
>  	RTE_ATOMIC(uint32_t) use;    /*usage counter */
> -	const struct rte_eth_rxtx_callback *cb;  /* callback handle */
> +	RTE_ATOMIC(const struct rte_eth_rxtx_callback *) cb;  /* callback handle
> */
>  	struct rte_bpf *bpf;
>  	struct rte_bpf_jit jit;
>  	/* used by control path only */
> @@ -80,9 +80,7 @@ static struct bpf_eth_cbh tx_cbh = {
>  static __rte_always_inline void
>  bpf_eth_cbi_inuse(struct bpf_eth_cbi *cbi)
>  {
> -	cbi->use++;
> -	/* make sure no store/load reordering could happen */
> -	rte_smp_mb();
> +	rte_atomic_fetch_add_explicit(&cbi->use, 1,
> rte_memory_order_acquire);
>  }
> 
>  /*
> @@ -91,9 +89,7 @@ bpf_eth_cbi_inuse(struct bpf_eth_cbi *cbi)
>  static __rte_always_inline void
>  bpf_eth_cbi_unuse(struct bpf_eth_cbi *cbi)
>  {
> -	/* make sure all previous loads are completed */
> -	rte_smp_rmb();
> -	cbi->use++;
> +	rte_atomic_fetch_add_explicit(&cbi->use, 1, rte_memory_order_release);
>  }
> 
>  /*
> @@ -104,15 +100,12 @@ bpf_eth_cbi_wait(const struct bpf_eth_cbi *cbi)
>  {
>  	uint32_t puse;
> 
> -	/* make sure all previous loads and stores are completed */
> -	rte_smp_mb();
> -
> -	puse = cbi->use;
> +	puse = rte_atomic_load_explicit(&cbi->use, rte_memory_order_acquire);
> 
>  	/* in use, busy wait till current RX/TX iteration is finished */
>  	if ((puse & BPF_ETH_CBI_INUSE) != 0) {
>  		RTE_WAIT_UNTIL_MASKED((__rte_atomic uint32_t
> *)(uintptr_t)&cbi->use,
> -			UINT32_MAX, !=, puse, rte_memory_order_relaxed);
> +			UINT32_MAX, !=, puse, rte_memory_order_acquire);
>  	}
>  }
> 
> @@ -201,11 +194,13 @@ bpf_rx_callback_vm(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
> 
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_vm(cbi->bpf, pkt, nb_pkts, 1) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -219,10 +214,12 @@ bpf_rx_callback_jit(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_jit(&cbi->jit, pkt, nb_pkts, 1) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -236,10 +233,12 @@ bpf_tx_callback_vm(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_vm(cbi->bpf, pkt, nb_pkts, 0) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -253,10 +252,12 @@ bpf_tx_callback_jit(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_jit(&cbi->jit, pkt, nb_pkts, 0) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -274,10 +275,12 @@ bpf_rx_callback_mb_vm(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_mb_vm(cbi->bpf, pkt, nb_pkts, 1) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -292,10 +295,12 @@ bpf_rx_callback_mb_jit(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_mb_jit(&cbi->jit, pkt, nb_pkts, 1) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -309,10 +314,12 @@ bpf_tx_callback_mb_vm(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_mb_vm(cbi->bpf, pkt, nb_pkts, 0) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -326,10 +333,12 @@ bpf_tx_callback_mb_jit(__rte_unused uint16_t port,
> __rte_unused uint16_t queue,
>  {
>  	struct bpf_eth_cbi *cbi;
>  	uint16_t rc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	cbi = user_param;
>  	bpf_eth_cbi_inuse(cbi);
> -	rc = (cbi->cb != NULL) ?
> +	cb = rte_atomic_load_explicit(&cbi->cb, rte_memory_order_acquire);
> +	rc = (cb != NULL) ?
>  		pkt_filter_mb_jit(&cbi->jit, pkt, nb_pkts, 0) :
>  		nb_pkts;
>  	bpf_eth_cbi_unuse(cbi);
> @@ -382,8 +391,7 @@ static void
>  bpf_eth_cbi_unload(struct bpf_eth_cbi *bc)
>  {
>  	/* mark this cbi as empty */
> -	bc->cb = NULL;
> -	rte_smp_mb();
> +	rte_atomic_store_explicit(&bc->cb, NULL, rte_memory_order_release);
> 
>  	/* make sure datapath doesn't use bpf anymore, then destroy bpf */
>  	bpf_eth_cbi_wait(bc);
> @@ -395,14 +403,17 @@ static void
>  bpf_eth_unload(struct bpf_eth_cbh *cbh, uint16_t port, uint16_t queue)
>  {
>  	struct bpf_eth_cbi *bc;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	bc = bpf_eth_cbh_find(cbh, port, queue);
> -	if (bc == NULL || bc->cb == NULL)
> +	if (bc == NULL)
> +		return;
> +	cb = rte_atomic_load_explicit(&bc->cb, rte_memory_order_acquire);
> +	if (cb == NULL)
>  		return;
> 
>  	if (cbh->type == BPF_ETH_RX)
> -		rte_eth_remove_rx_callback(port, queue, bc->cb);
> +		rte_eth_remove_rx_callback(port, queue, cb);
>  	else
> -		rte_eth_remove_tx_callback(port, queue, bc->cb);
> +		rte_eth_remove_tx_callback(port, queue, cb);
> 
>  	bpf_eth_cbi_unload(bc);
>  }
> @@ -439,6 +450,7 @@ bpf_eth_elf_install(struct bpf_eth_cbh *cbh, uint16_t
> port, uint16_t queue,
>  	rte_rx_callback_fn frx;
>  	rte_tx_callback_fn ftx;
>  	struct rte_bpf_jit jit;
> +	const struct rte_eth_rxtx_callback *cb;
> 
>  	frx = NULL;
>  	ftx = NULL;
> @@ -470,17 +482,18 @@ bpf_eth_elf_install(struct bpf_eth_cbh *cbh, uint16_t
> port, uint16_t queue,
>  		return -ENOMEM;
> 
>  	/* remove old one, if any */
> -	if (bc->cb != NULL)
> +	if (rte_atomic_load_explicit(&bc->cb, rte_memory_order_acquire) !=
> NULL)
>  		bpf_eth_unload(cbh, port, queue);
> 
>  	bc->bpf = bpf;
>  	bc->jit = jit;
> 
>  	if (cbh->type == BPF_ETH_RX)
> -		bc->cb = rte_eth_add_rx_callback(port, queue, frx, bc);
> +		cb = rte_eth_add_rx_callback(port, queue, frx, bc);
>  	else
> -		bc->cb = rte_eth_add_tx_callback(port, queue, ftx, bc);
> +		cb = rte_eth_add_tx_callback(port, queue, ftx, bc);
> 
> -	if (bc->cb == NULL) {
> +	rte_atomic_store_explicit(&bc->cb, cb, rte_memory_order_release);
> +	if (cb == NULL) {
>  		rc = -rte_errno;
>  		bpf_eth_cbi_cleanup(bc);