>
> 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);
