Gemini's suggestion FWIW (looks sensible but not tested):
---
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.
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.
---
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);
