This is AI remix of the smp_mb barrier removal patch, the patch series from
Marat.
The Rx/Tx callback use counter assumes only one thread at a time
does Rx/Tx burst on a queue. That is not true for Tx with
RTE_ETH_TX_OFFLOAD_MT_LOCKFREE. With two threads in the callback
the counter can be even while the program is still running, so
unload can destroy a BPF program that is in use. Without
enable_stdatomic the increment is not atomic and an update can be
lost, so unload can also wait forever.
Replace the odd/even counter with an atomic count of the threads
inside the callback. The datapath increments it on entry and
decrements it with release on exit. Unload clears cb and waits for
zero with acquire.
Make cb atomic. The datapath increment and cb load, and the unload
cb store and count load, are seq_cst; this replaces rte_smp_mb()
and rte_smp_rmb(). Install publishes cb with release so the
datapath sees bpf and jit.
On x86 the datapath is now two locked instructions per burst
instead of one.
Add a test that loads and unloads a Tx filter while worker lcores
transmit on the same MT_LOCKFREE queue of a null device.
Bugzilla ID: 2051
Fixes: a93ff62a8938 ("bpf: introduce basic Rx/Tx filters")
Cc: [email protected]
Signed-off-by: Stephen Hemminger <[email protected]>
---
app/test/test_bpf.c | 148 +++++++++++++++++++++++++++++++-
lib/bpf/bpf_pkt.c | 177 +++++++++++++++++----------------------
lib/bpf/rte_bpf_ethdev.h | 2 +
3 files changed, 225 insertions(+), 102 deletions(-)
diff --git a/app/test/test_bpf.c b/app/test/test_bpf.c
index 3205afaa63..f21e906c6f 100644
--- a/app/test/test_bpf.c
+++ b/app/test/test_bpf.c
@@ -4329,6 +4329,10 @@ test_bpf_elf_load(load_elf_image_t load_elf_image)
#include <rte_ethdev.h>
#include <rte_bpf_ethdev.h>
#include <rte_bus_vdev.h>
+#include <rte_cycles.h>
+#include <rte_launch.h>
+#include <rte_lcore.h>
+#include <rte_pause.h>
#include "test_bpf_filter.h"
@@ -4336,7 +4340,8 @@ test_bpf_elf_load(load_elf_image_t load_elf_image)
#define BPF_TEST_POOLSIZE 256 /* at least 2x burst */
#define BPF_TEST_PKT_LEN 64 /* Ether + IP + TCP */
-static int null_vdev_setup(const char *name, uint16_t *port, struct
rte_mempool *pool)
+static int null_vdev_setup(const char *name, uint16_t *port, struct
rte_mempool *pool,
+ uint64_t tx_offloads)
{
int ret;
@@ -4347,7 +4352,9 @@ static int null_vdev_setup(const char *name, uint16_t
*port, struct rte_mempool
ret = rte_eth_dev_get_port_by_name(name, port);
TEST_ASSERT(ret == 0, "failed to get port id for %s: %d", name, ret);
- struct rte_eth_conf conf = { };
+ struct rte_eth_conf conf = {
+ .txmode.offloads = tx_offloads,
+ };
ret = rte_eth_dev_configure(*port, 1, 1, &conf);
TEST_ASSERT(ret == 0, "failed to configure port %u: %d", *port, ret);
@@ -4488,7 +4495,7 @@ test_bpf_elf_tx_load(load_elf_image_t load_elf_image)
SOCKET_ID_ANY);
TEST_ASSERT(mb_pool != NULL, "failed to create mempool");
- ret = null_vdev_setup(null_dev, &port, mb_pool);
+ ret = null_vdev_setup(null_dev, &port, mb_pool, 0);
if (ret != 0)
goto fail;
@@ -4577,7 +4584,7 @@ test_bpf_elf_rx_load(load_elf_image_t load_elf_image)
SOCKET_ID_ANY);
TEST_ASSERT(pool != NULL, "failed to create mempool");
- ret = null_vdev_setup(null_dev, &port, pool);
+ ret = null_vdev_setup(null_dev, &port, pool, 0);
if (ret != 0)
goto fail;
@@ -4618,6 +4625,136 @@ test_bpf_elf_rx_load(load_elf_image_t load_elf_image)
}
+#define BPF_MT_BURST 32
+#define BPF_MT_POOLSIZE 4096
+#define BPF_MT_ITERATIONS 1000
+
+static RTE_ATOMIC(bool) bpf_mt_stop;
+
+struct bpf_mt_arg {
+ uint16_t port;
+ struct rte_mempool *pool;
+};
+
+/* Transmit on the same queue as the other workers until told to stop. */
+static int
+bpf_mt_tx_worker(void *arg)
+{
+ const struct bpf_mt_arg *mt = arg;
+ struct rte_mbuf *pkts[BPF_MT_BURST];
+
+ while (!rte_atomic_load_explicit(&bpf_mt_stop,
rte_memory_order_relaxed)) {
+ if (rte_pktmbuf_alloc_bulk(mt->pool, pkts, BPF_MT_BURST) != 0) {
+ rte_pause();
+ continue;
+ }
+ setup_mbufs(pkts, BPF_MT_BURST);
+
+ uint16_t sent = rte_eth_tx_burst(mt->port, 0, pkts,
BPF_MT_BURST);
+ rte_pktmbuf_free_bulk(pkts + sent, BPF_MT_BURST - sent);
+ }
+ return 0;
+}
+
+/*
+ * Load and unload a Tx filter while several lcores do tx burst
+ * on the same MT_LOCKFREE queue.
+ */
+static int
+test_bpf_elf_tx_mt_lockfree(load_elf_image_t load_elf_image)
+{
+ static const struct rte_bpf_arg prog_arg = {
+ .type = RTE_BPF_ARG_PTR,
+ .size = sizeof(struct dummy_net),
+ };
+ static const char null_dev[] = "net_null_bpf_mt";
+ struct bpf_mt_arg mt = { .port = UINT16_MAX };
+ struct rte_eth_dev_info dev_info;
+ unsigned int lcore, workers = 0;
+ int ret;
+
+ RTE_LCORE_FOREACH_WORKER(lcore)
+ workers++;
+ if (workers < 2) {
+ printf("%s: need at least 2 worker lcores, skipping\n",
__func__);
+ return TEST_SKIPPED;
+ }
+
+ mt.pool = rte_pktmbuf_pool_create("bpf_mt_test_pool", BPF_MT_POOLSIZE,
+ 0, 0, RTE_MBUF_DEFAULT_BUF_SIZE,
+ SOCKET_ID_ANY);
+ TEST_ASSERT_NOT_NULL(mt.pool, "failed to create mempool");
+
+ ret = null_vdev_setup(null_dev, &mt.port, mt.pool,
RTE_ETH_TX_OFFLOAD_MT_LOCKFREE);
+ if (ret != 0)
+ goto fail;
+
+ ret = rte_eth_dev_info_get(mt.port, &dev_info);
+ if (ret != 0)
+ goto fail;
+ if (!(dev_info.tx_offload_capa & RTE_ETH_TX_OFFLOAD_MT_LOCKFREE)) {
+ ret = TEST_SKIPPED;
+ goto fail;
+ }
+
+ rte_atomic_store_explicit(&bpf_mt_stop, false,
rte_memory_order_relaxed);
+ RTE_LCORE_FOREACH_WORKER(lcore)
+ rte_eal_remote_launch(bpf_mt_tx_worker, &mt, lcore);
+
+ for (unsigned int i = 0; i < BPF_MT_ITERATIONS; i++) {
+ uint32_t flags = (i & 1) ? RTE_BPF_ETH_F_JIT : 0;
+ struct rte_bpf *bpf;
+ struct rte_bpf_jit jit;
+
+ bpf = load_elf_image(app_test_bpf_filter_o,
app_test_bpf_filter_o_len,
+ "filter", NULL, 0, &prog_arg);
+ if (bpf == NULL) {
+ printf("%s@%d: failed to load BPF filter\n", __func__,
__LINE__);
+ ret = TEST_FAILED;
+ break;
+ }
+
+ rte_bpf_get_jit(bpf, &jit);
+ if (jit.func == NULL)
+ flags = 0;
+
+ ret = rte_bpf_eth_tx_install(mt.port, 0, bpf, flags);
+ if (ret != 0) {
+ printf("%s@%d: install failed: %d\n", __func__,
__LINE__, ret);
+ rte_bpf_destroy(bpf);
+ ret = TEST_FAILED;
+ break;
+ }
+
+ rte_delay_us(10);
+ rte_bpf_eth_tx_unload(mt.port, 0);
+ }
+
+ rte_atomic_store_explicit(&bpf_mt_stop, true, rte_memory_order_relaxed);
+ rte_eal_mp_wait_lcore();
+
+ if (ret == 0) {
+ unsigned int avail = rte_mempool_avail_count(mt.pool);
+
+ if (avail != BPF_MT_POOLSIZE) {
+ printf("%s: mempool available %u != %u\n",
+ __func__, avail, BPF_MT_POOLSIZE);
+ ret = TEST_FAILED;
+ }
+ }
+
+fail:
+ if (mt.port != UINT16_MAX) {
+ rte_eth_dev_stop(mt.port);
+ rte_vdev_uninit(null_dev);
+ }
+ rte_mempool_free(mt.pool);
+
+ if (ret == TEST_SKIPPED)
+ return TEST_SKIPPED;
+ return ret == 0 ? TEST_SUCCESS : TEST_FAILED;
+}
+
static int
test_bpf_elf(void)
{
@@ -4637,6 +4774,9 @@ test_bpf_elf(void)
ret = test_bpf_elf_rx_load(elf_image_loaders[li]);
}
+ if (ret == TEST_SUCCESS)
+ ret = test_bpf_elf_tx_mt_lockfree(load_elf_image_direct);
+
return ret;
}
diff --git a/lib/bpf/bpf_pkt.c b/lib/bpf/bpf_pkt.c
index f072fdaaed..e9fe4aeb5a 100644
--- a/lib/bpf/bpf_pkt.c
+++ b/lib/bpf/bpf_pkt.c
@@ -26,8 +26,8 @@
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(uint32_t) use; /* number of datapath threads in callback
*/
+ 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 */
@@ -36,12 +36,6 @@ struct __rte_cache_aligned bpf_eth_cbi {
uint16_t queue;
};
-/*
- * Odd number means that callback is used by datapath.
- * Even number means that callback is not used by datapath.
- */
-#define BPF_ETH_CBI_INUSE 1
-
/*
* List to manage RX/TX installed callbacks.
*/
@@ -75,45 +69,50 @@ static struct bpf_eth_cbh tx_cbh = {
};
/*
- * Marks given callback as used by datapath.
+ * Removing an rx/tx callback is done in two steps (similar to RCU).
+ * The callback is removed from the ethdev queue so later bursts do not
+ * call it, but a burst already in progress may still be running it.
+ *
+ * The use counter is the number of threads inside the callback.
+ * There can be more than one on a Tx queue with MT_LOCKFREE.
+ *
+ * Datapath increments use then loads cb; unload stores cb then loads use.
+ * These are seq_cst, so either the datapath sees cb == NULL and skips
+ * the program, or unload sees use != 0 and waits for zero. The release
+ * decrement pairs with the acquire wait, so the program is destroyed
+ * only after its last use.
+ */
+
+/*
+ * Enter the callback. Returns true if the BPF program may be run.
*/
-static __rte_always_inline void
-bpf_eth_cbi_inuse(struct bpf_eth_cbi *cbi)
+static inline bool
+bpf_eth_cbi_enter(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_seq_cst);
+ return rte_atomic_load_explicit(&cbi->cb, rte_memory_order_seq_cst) !=
NULL;
}
/*
- * Marks given callback list as not used by datapath.
+ * Leave the callback.
*/
-static __rte_always_inline void
-bpf_eth_cbi_unuse(struct bpf_eth_cbi *cbi)
+static inline void
+bpf_eth_cbi_exit(struct bpf_eth_cbi *cbi)
{
- /* make sure all previous loads are completed */
- rte_smp_rmb();
- cbi->use++;
+ rte_atomic_fetch_sub_explicit(&cbi->use, 1, rte_memory_order_release);
}
/*
* Waits till datapath finished using given callback.
+ * Must be called after cb is cleared.
*/
static void
bpf_eth_cbi_wait(const struct bpf_eth_cbi *cbi)
{
- uint32_t puse;
+ __rte_atomic uint32_t *use = (__rte_atomic uint32_t
*)(uintptr_t)&cbi->use;
- /* make sure all previous loads and stores are completed */
- rte_smp_mb();
-
- puse = cbi->use;
-
- /* 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);
- }
+ if (rte_atomic_load_explicit(use, rte_memory_order_seq_cst) != 0)
+ RTE_WAIT_UNTIL_MASKED(use, UINT32_MAX, ==, 0,
rte_memory_order_acquire);
}
static void
@@ -262,16 +261,12 @@ bpf_rx_callback_vm(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
__rte_unused uint16_t max_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
-
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_vm(cbi->bpf, pkt, nb_pkts, 1) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_vm(cbi->bpf, pkt, nb_pkts, 1);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
static uint16_t
@@ -280,15 +275,12 @@ bpf_rx_callback_jit(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
__rte_unused uint16_t max_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_jit(&cbi->jit, pkt, nb_pkts, 1) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_jit(&cbi->jit, pkt, nb_pkts, 1);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
static uint16_t
@@ -296,15 +288,12 @@ bpf_tx_callback_vm(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
struct rte_mbuf *pkt[], uint16_t nb_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_vm(cbi->bpf, pkt, nb_pkts, 0) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_vm(cbi->bpf, pkt, nb_pkts, 0);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
static uint16_t
@@ -312,15 +301,12 @@ bpf_tx_callback_jit(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
struct rte_mbuf *pkt[], uint16_t nb_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_jit(&cbi->jit, pkt, nb_pkts, 0) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_jit(&cbi->jit, pkt, nb_pkts, 0);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
/*
@@ -333,15 +319,12 @@ bpf_rx_callback_mb_vm(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
__rte_unused uint16_t max_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_mb_vm(cbi->bpf, pkt, nb_pkts, 1) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_mb_vm(cbi->bpf, pkt, nb_pkts, 1);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
static uint16_t
@@ -350,15 +333,12 @@ bpf_rx_callback_mb_jit(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
__rte_unused uint16_t max_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_mb_jit(&cbi->jit, pkt, nb_pkts, 1) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_mb_jit(&cbi->jit, pkt, nb_pkts, 1);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
static uint16_t
@@ -366,15 +346,12 @@ bpf_tx_callback_mb_vm(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
struct rte_mbuf *pkt[], uint16_t nb_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_mb_vm(cbi->bpf, pkt, nb_pkts, 0) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_mb_vm(cbi->bpf, pkt, nb_pkts, 0);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
static uint16_t
@@ -382,15 +359,12 @@ bpf_tx_callback_mb_jit(__rte_unused uint16_t port,
__rte_unused uint16_t queue,
struct rte_mbuf *pkt[], uint16_t nb_pkts, void *user_param)
{
struct bpf_eth_cbi *cbi;
- uint16_t rc;
cbi = user_param;
- bpf_eth_cbi_inuse(cbi);
- rc = (cbi->cb != NULL) ?
- pkt_filter_mb_jit(&cbi->jit, pkt, nb_pkts, 0) :
- nb_pkts;
- bpf_eth_cbi_unuse(cbi);
- return rc;
+ if (bpf_eth_cbi_enter(cbi))
+ nb_pkts = pkt_filter_mb_jit(&cbi->jit, pkt, nb_pkts, 0);
+ bpf_eth_cbi_exit(cbi);
+ return nb_pkts;
}
static rte_rx_callback_fn
@@ -438,8 +412,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_seq_cst);
/* make sure datapath doesn't use bpf anymore, then destroy bpf */
bpf_eth_cbi_wait(bc);
@@ -450,16 +423,21 @@ bpf_eth_cbi_unload(struct bpf_eth_cbi *bc)
static void
bpf_eth_unload(struct bpf_eth_cbh *cbh, uint16_t port, uint16_t queue)
{
+ const struct rte_eth_rxtx_callback *cb;
struct bpf_eth_cbi *bc;
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_relaxed);
+ 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);
}
@@ -494,6 +472,7 @@ bpf_eth_elf_install(struct bpf_eth_cbh *cbh, uint16_t port,
uint16_t queue,
struct rte_bpf *bpf, uint32_t flags)
{
int32_t rc;
+ const struct rte_eth_rxtx_callback *cb;
struct bpf_eth_cbi *bc;
rte_rx_callback_fn frx;
rte_tx_callback_fn ftx;
@@ -534,18 +513,20 @@ 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)
- bpf_eth_unload(cbh, port, queue);
+ 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);
+
+ /* publish bpf and jit to the datapath */
+ rte_atomic_store_explicit(&bc->cb, cb, rte_memory_order_release);
- if (bc->cb == NULL) {
+ if (cb == NULL) {
rc = -rte_errno;
bpf_eth_cbi_cleanup(bc);
} else
diff --git a/lib/bpf/rte_bpf_ethdev.h b/lib/bpf/rte_bpf_ethdev.h
index 686fda8d94..9f14a624d4 100644
--- a/lib/bpf/rte_bpf_ethdev.h
+++ b/lib/bpf/rte_bpf_ethdev.h
@@ -14,6 +14,8 @@
* same port/queue from different threads in parallel.
* - though it allows load/unload at runtime
* (while RX/TX is ongoing on given port/queue).
+ * - TX burst on the same port/queue may be called from more than
+ * one thread when RTE_ETH_TX_OFFLOAD_MT_LOCKFREE is enabled.
* - allows only one BPF program per port/queue,
* i.e. new load will replace previously loaded for that port/queue BPF
program.
* Filter behaviour - if BPF program returns zero value for a given packet,
--
2.53.0
