[PATCH] fib: implement RCU rule reclamation

[PATCH] fib: implement RCU rule reclamation

[Vladimir Medvedkin]

Currently, for DIR24-8 algorithm, the tbl8 group is freed even though the
readers might be using the tbl8 group entries. The freed tbl8 group can
be reallocated quickly. As a result, lookup may be performed incorrectly.

To address that, RCU QSBR is integrated for safe tbl8 group reclamation.

Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
---
 lib/fib/dir24_8.c   | 104 +++++++++++++++++++++++++++++++++++++++-----
 lib/fib/dir24_8.h   |   9 ++++
 lib/fib/meson.build |   1 +
 lib/fib/rte_fib.c   |  11 +++++
 lib/fib/rte_fib.h   |  50 ++++++++++++++++++++-
 lib/fib/version.map |   7 +++
 6 files changed, 171 insertions(+), 11 deletions(-)

diff --git a/lib/fib/dir24_8.c b/lib/fib/dir24_8.c
index c739e92304..f884b02d2c 100644
--- a/lib/fib/dir24_8.c
+++ b/lib/fib/dir24_8.c
@@ -14,6 +14,7 @@
 #include <rte_rib.h>
 #include <rte_fib.h>
 #include "dir24_8.h"
+#include "fib_log.h"
 
 #ifdef CC_DIR24_8_AVX512_SUPPORT
 
@@ -176,6 +177,13 @@ tbl8_alloc(struct dir24_8_tbl *dp, uint64_t nh)
 	uint8_t	*tbl8_ptr;
 
 	tbl8_idx = tbl8_get_idx(dp);
+	if ((tbl8_idx == -ENOSPC) && dp->dq != NULL) {
+		/* If there are no tbl8 groups try to reclaim one. */
+		if (rte_rcu_qsbr_dq_reclaim(dp->dq, 1,
+				NULL, NULL, NULL) == 0)
+			tbl8_idx = tbl8_get_idx(dp);
+	}
+
 	if (tbl8_idx < 0)
 		return tbl8_idx;
 	tbl8_ptr = (uint8_t *)dp->tbl8 +
@@ -189,6 +197,27 @@ tbl8_alloc(struct dir24_8_tbl *dp, uint64_t nh)
 	return tbl8_idx;
 }
 
+static void
+tbl8_cleanup_and_free(struct dir24_8_tbl *dp, uint64_t tbl8_idx)
+{
+	uint8_t *ptr = (uint8_t *)dp->tbl8 +
+		(tbl8_idx * DIR24_8_TBL8_GRP_NUM_ENT << dp->nh_sz);
+
+	memset(ptr, 0, DIR24_8_TBL8_GRP_NUM_ENT << dp->nh_sz);
+	tbl8_free_idx(dp, tbl8_idx);
+	dp->cur_tbl8s--;
+}
+
+static void
+__rcu_qsbr_free_resource(void *p, void *data, unsigned int n)
+{
+	struct dir24_8_tbl *dp = p;
+	uint64_t tbl8_idx = *(uint64_t *)data;
+	RTE_SET_USED(n);
+
+	tbl8_cleanup_and_free(dp, tbl8_idx);
+}
+
 static void
 tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 {
@@ -210,8 +239,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint8_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr8[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_2B:
 		ptr16 = &((uint16_t *)dp->tbl8)[tbl8_idx *
@@ -223,8 +250,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint16_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr16[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_4B:
 		ptr32 = &((uint32_t *)dp->tbl8)[tbl8_idx *
@@ -236,8 +261,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint32_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr32[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_8B:
 		ptr64 = &((uint64_t *)dp->tbl8)[tbl8_idx *
@@ -249,12 +272,20 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint64_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr64[i] = 0;
 		break;
 	}
-	tbl8_free_idx(dp, tbl8_idx);
-	dp->cur_tbl8s--;
+
+	if (dp->v == NULL)
+		tbl8_cleanup_and_free(dp, tbl8_idx);
+	else if (dp->rcu_mode == RTE_FIB_QSBR_MODE_SYNC) {
+		rte_rcu_qsbr_synchronize(dp->v,
+			RTE_QSBR_THRID_INVALID);
+		tbl8_cleanup_and_free(dp, tbl8_idx);
+	} else { /* RTE_FIB_QSBR_MODE_DQ */
+		if (rte_rcu_qsbr_dq_enqueue(dp->dq,
+				(void *)&tbl8_idx))
+			FIB_LOG(ERR, "Failed to push QSBR FIFO");
+	}
 }
 
 static int
@@ -569,7 +600,60 @@ dir24_8_free(void *p)
 {
 	struct dir24_8_tbl *dp = (struct dir24_8_tbl *)p;
 
+	if (dp->dq != NULL)
+		rte_rcu_qsbr_dq_delete(dp->dq);
+
 	rte_free(dp->tbl8_idxes);
 	rte_free(dp->tbl8);
 	rte_free(dp);
 }
+
+int
+dir24_8_rcu_qsbr_add(struct dir24_8_tbl *dp, struct rte_fib_rcu_config *cfg,
+	const char *name)
+{
+	struct rte_rcu_qsbr_dq_parameters params = {0};
+	char rcu_dq_name[RTE_RCU_QSBR_DQ_NAMESIZE];
+
+	if (dp == NULL || cfg == NULL) {
+		rte_errno = EINVAL;
+		return 1;
+	}
+
+	if (dp->v != NULL) {
+		rte_errno = EEXIST;
+		return 1;
+	}
+
+	if (cfg->mode == RTE_FIB_QSBR_MODE_SYNC) {
+		/* No other things to do. */
+	} else if (cfg->mode == RTE_FIB_QSBR_MODE_DQ) {
+		/* Init QSBR defer queue. */
+		snprintf(rcu_dq_name, sizeof(rcu_dq_name),
+				"FIB_RCU_%s", name);
+		params.name = rcu_dq_name;
+		params.size = cfg->dq_size;
+		if (params.size == 0)
+			params.size = RTE_FIB_RCU_DQ_RECLAIM_SZ;
+		params.trigger_reclaim_limit = cfg->reclaim_thd;
+		params.max_reclaim_size = cfg->reclaim_max;
+		if (params.max_reclaim_size == 0)
+			params.max_reclaim_size = RTE_FIB_RCU_DQ_RECLAIM_MAX;
+		params.esize = sizeof(uint64_t);
+		params.free_fn = __rcu_qsbr_free_resource;
+		params.p = dp;
+		params.v = cfg->v;
+		dp->dq = rte_rcu_qsbr_dq_create(&params);
+		if (dp->dq == NULL) {
+			FIB_LOG(ERR, "LPM defer queue creation failed");
+			return 1;
+		}
+	} else {
+		rte_errno = EINVAL;
+		return 1;
+	}
+	dp->rcu_mode = cfg->mode;
+	dp->v = cfg->v;
+
+	return 0;
+}
\ No newline at end of file
diff --git a/lib/fib/dir24_8.h b/lib/fib/dir24_8.h
index 7125049f15..08fd818ce4 100644
--- a/lib/fib/dir24_8.h
+++ b/lib/fib/dir24_8.h
@@ -10,6 +10,7 @@
 
 #include <rte_prefetch.h>
 #include <rte_branch_prediction.h>
+#include <rte_rcu_qsbr.h>
 
 /**
  * @file
@@ -30,6 +31,10 @@ struct dir24_8_tbl {
 	uint32_t	rsvd_tbl8s;	/**< Number of reserved tbl8s */
 	uint32_t	cur_tbl8s;	/**< Current number of tbl8s */
 	enum rte_fib_dir24_8_nh_sz	nh_sz;	/**< Size of nexthop entry */
+	/* RCU config. */
+	enum rte_fib_qsbr_mode rcu_mode;/* Blocking, defer queue. */
+	struct rte_rcu_qsbr *v;		/* RCU QSBR variable. */
+	struct rte_rcu_qsbr_dq *dq;	/* RCU QSBR defer queue. */
 	uint64_t	def_nh;		/**< Default next hop */
 	uint64_t	*tbl8;		/**< tbl8 table. */
 	uint64_t	*tbl8_idxes;	/**< bitmap containing free tbl8 idxes*/
@@ -250,4 +255,8 @@ int
 dir24_8_modify(struct rte_fib *fib, uint32_t ip, uint8_t depth,
 	uint64_t next_hop, int op);
 
+int
+dir24_8_rcu_qsbr_add(struct dir24_8_tbl *dp, struct rte_fib_rcu_config *cfg,
+	const char *name);
+
 #endif /* _DIR24_8_H_ */
diff --git a/lib/fib/meson.build b/lib/fib/meson.build
index 6795f41a0a..1895f37050 100644
--- a/lib/fib/meson.build
+++ b/lib/fib/meson.build
@@ -11,6 +11,7 @@ endif
 sources = files('rte_fib.c', 'rte_fib6.c', 'dir24_8.c', 'trie.c')
 headers = files('rte_fib.h', 'rte_fib6.h')
 deps += ['rib']
+deps += ['rcu']
 
 # compile AVX512 version if:
 # we are building 64-bit binary AND binutils can generate proper code
diff --git a/lib/fib/rte_fib.c b/lib/fib/rte_fib.c
index 4f9fba5a4f..f1b73d64cb 100644
--- a/lib/fib/rte_fib.c
+++ b/lib/fib/rte_fib.c
@@ -338,3 +338,14 @@ rte_fib_select_lookup(struct rte_fib *fib,
 		return -EINVAL;
 	}
 }
+
+int
+rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg)
+{
+	switch (fib->type) {
+	case RTE_FIB_DIR24_8:
+	        return dir24_8_rcu_qsbr_add(fib->dp, cfg, fib->name);
+	default:
+	        return -ENOTSUP;
+	}
+}
\ No newline at end of file
diff --git a/lib/fib/rte_fib.h b/lib/fib/rte_fib.h
index d7a5aafe53..346eb7f149 100644
--- a/lib/fib/rte_fib.h
+++ b/lib/fib/rte_fib.h
@@ -16,7 +16,7 @@
  */
 
 #include <stdint.h>
-
+#include <rte_rcu_qsbr.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -28,6 +28,19 @@ struct rte_rib;
 /** Maximum depth value possible for IPv4 FIB. */
 #define RTE_FIB_MAXDEPTH	32
 
+/** @internal Default RCU defer queue entries to reclaim in one go. */
+#define RTE_FIB_RCU_DQ_RECLAIM_MAX	16
+/** @internal Default RCU defer queue size. */
+#define RTE_FIB_RCU_DQ_RECLAIM_SZ	128
+
+/** RCU reclamation modes */
+enum rte_fib_qsbr_mode {
+	/** Create defer queue for reclaim. */
+	RTE_FIB_QSBR_MODE_DQ = 0,
+	/** Use blocking mode reclaim. No defer queue created. */
+	RTE_FIB_QSBR_MODE_SYNC
+};
+
 /** Type of FIB struct */
 enum rte_fib_type {
 	RTE_FIB_DUMMY,		/**< RIB tree based FIB */
@@ -89,6 +102,22 @@ struct rte_fib_conf {
 	};
 };
 
+/** FIB RCU QSBR configuration structure. */
+struct rte_fib_rcu_config {
+	struct rte_rcu_qsbr *v;	/* RCU QSBR variable. */
+	/* Mode of RCU QSBR. RTE_FIB_QSBR_MODE_xxx
+	 * '0' for default: create defer queue for reclaim.
+	 */
+	enum rte_fib_qsbr_mode mode;
+	uint32_t dq_size;	/* RCU defer queue size.
+				 * default: RTE_FIB_RCU_DQ_RECLAIM_SZ.
+				 */
+	uint32_t reclaim_thd;	/* Threshold to trigger auto reclaim. */
+	uint32_t reclaim_max;	/* Max entries to reclaim in one go.
+				 * default: RTE_FIB_RCU_DQ_RECLAIM_MAX.
+				 */
+};
+
 /**
  * Create FIB
  *
@@ -219,6 +248,25 @@ rte_fib_get_rib(struct rte_fib *fib);
 int
 rte_fib_select_lookup(struct rte_fib *fib, enum rte_fib_lookup_type type);
 
+/**
+ * Associate RCU QSBR variable with a FIB object.
+ *
+ * @param fib
+ *   the fib object to add RCU QSBR
+ * @param cfg
+ *   RCU QSBR configuration
+ * @return
+ *   On success - 0
+ *   On error - 1 with error code set in rte_errno.
+ *   Possible rte_errno codes are:
+ *   - EINVAL - invalid pointer
+ *   - EEXIST - already added QSBR
+ *   - ENOMEM - memory allocation failure
+ *   - ENOTSUP - not supported by configured dataplane algorithm
+ */
+__rte_experimental
+int rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/fib/version.map b/lib/fib/version.map
index c6d2769611..df8f113df3 100644
--- a/lib/fib/version.map
+++ b/lib/fib/version.map
@@ -22,3 +22,10 @@ DPDK_25 {
 
 	local: *;
 };
+
+EXPERIMENTAL {
+	global:
+
+	# added in 24.11
+	rte_fib_rcu_qsbr_add;
+};
-- 
2.34.1

Re: [PATCH] fib: implement RCU rule reclamation

[Robin Jarry]

Vladimir Medvedkin, Sep 06, 2024 at 13:09:
> Currently, for DIR24-8 algorithm, the tbl8 group is freed even though the
> readers might be using the tbl8 group entries. The freed tbl8 group can
> be reallocated quickly. As a result, lookup may be performed incorrectly.
>
> To address that, RCU QSBR is integrated for safe tbl8 group reclamation.
>
> Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
> ---

> diff --git a/lib/fib/meson.build b/lib/fib/meson.build
> index 6795f41a0a..1895f37050 100644
> --- a/lib/fib/meson.build
> +++ b/lib/fib/meson.build
> @@ -11,6 +11,7 @@ endif
>  sources = files('rte_fib.c', 'rte_fib6.c', 'dir24_8.c', 'trie.c')
>  headers = files('rte_fib.h', 'rte_fib6.h')
>  deps += ['rib']
> +deps += ['rcu']

Hi Vladimir,

thanks a lot for working on this!

I tested with static linking and there is a missing dependency to 
static_rte_rcu:

In file included from ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:6:
../subprojects/dpdk/lib/fib/rte_fib.h:19:10: fatal error: rte_rcu_qsbr.h: No such file or directory
   19 | #include <rte_rcu_qsbr.h>
      |          ^~~~~~~~~~~~~~~~

After adding it:

@@ -45,7 +45,7 @@ if dpdk_conf.has('RTE_ARCH_X86_64') and binutils_ok
     elif cc.has_multi_arguments('-mavx512f', '-mavx512dq')
         dir24_8_avx512_tmp = static_library('dir24_8_avx512_tmp',
                 'dir24_8_avx512.c',
-                dependencies: static_rte_eal,
+                dependencies: [static_rte_eal, static_rte_rcu],
                 c_args: cflags + ['-mavx512f', '-mavx512dq'])
         objs += dir24_8_avx512_tmp.extract_objects('dir24_8_avx512.c')
         cflags += ['-DCC_DIR24_8_AVX512_SUPPORT']

I get another error:

In file included from /usr/lib/gcc/x86_64-redhat-linux/14/include/immintrin.h:65,
                 from /usr/lib/gcc/x86_64-redhat-linux/14/include/x86intrin.h:32,
                 from ../subprojects/dpdk/lib/eal/x86/include/rte_vect.h:26,
                 from ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:5:
/usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h: In function ‘dir24_8_vec_lookup_x16’:
/usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h:1947:1: error: inlining failed in call to ‘always_inline’ ‘_mm512_shuffle_epi8’: target specific option mismatch
 1947 | _mm512_shuffle_epi8 (__m512i __A, __m512i __B)
      | ^~~~~~~~~~~~~~~~~~~
../subprojects/dpdk/lib/fib/dir24_8_avx512.c:38:26: note: called from here
   38 |                 ip_vec = _mm512_shuffle_epi8(ip_vec, bswap32);
      |                          ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
/usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h:1947:1: error: inlining failed in call to ‘always_inline’ ‘_mm512_shuffle_epi8’: target specific option mismatch
 1947 | _mm512_shuffle_epi8 (__m512i __A, __m512i __B)
      | ^~~~~~~~~~~~~~~~~~~
../subprojects/dpdk/lib/fib/dir24_8_avx512.c:38:26: note: called from here
   38 |                 ip_vec = _mm512_shuffle_epi8(ip_vec, bswap32);
      |                          ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

I'm not sure what to do at this point.

Re: [PATCH] fib: implement RCU rule reclamation

[David Marchand]

On Fri, Sep 27, 2024 at 6:13 PM Robin Jarry <rjarry@redhat.com> wrote:
>
> Vladimir Medvedkin, Sep 06, 2024 at 13:09:
> > Currently, for DIR24-8 algorithm, the tbl8 group is freed even though the
> > readers might be using the tbl8 group entries. The freed tbl8 group can
> > be reallocated quickly. As a result, lookup may be performed incorrectly.
> >
> > To address that, RCU QSBR is integrated for safe tbl8 group reclamation.
> >
> > Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
> > ---
>
> > diff --git a/lib/fib/meson.build b/lib/fib/meson.build
> > index 6795f41a0a..1895f37050 100644
> > --- a/lib/fib/meson.build
> > +++ b/lib/fib/meson.build
> > @@ -11,6 +11,7 @@ endif
> >  sources = files('rte_fib.c', 'rte_fib6.c', 'dir24_8.c', 'trie.c')
> >  headers = files('rte_fib.h', 'rte_fib6.h')
> >  deps += ['rib']
> > +deps += ['rcu']
>
> Hi Vladimir,
>
> thanks a lot for working on this!
>
> I tested with static linking and there is a missing dependency to
> static_rte_rcu:
>
> In file included from ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:6:
> ../subprojects/dpdk/lib/fib/rte_fib.h:19:10: fatal error: rte_rcu_qsbr.h: No such file or directory
>    19 | #include <rte_rcu_qsbr.h>
>       |          ^~~~~~~~~~~~~~~~
>
> After adding it:
>
> @@ -45,7 +45,7 @@ if dpdk_conf.has('RTE_ARCH_X86_64') and binutils_ok
>      elif cc.has_multi_arguments('-mavx512f', '-mavx512dq')
>          dir24_8_avx512_tmp = static_library('dir24_8_avx512_tmp',
>                  'dir24_8_avx512.c',
> -                dependencies: static_rte_eal,
> +                dependencies: [static_rte_eal, static_rte_rcu],
>                  c_args: cflags + ['-mavx512f', '-mavx512dq'])
>          objs += dir24_8_avx512_tmp.extract_objects('dir24_8_avx512.c')
>          cflags += ['-DCC_DIR24_8_AVX512_SUPPORT']
>
> I get another error:
>
> In file included from /usr/lib/gcc/x86_64-redhat-linux/14/include/immintrin.h:65,
>                  from /usr/lib/gcc/x86_64-redhat-linux/14/include/x86intrin.h:32,
>                  from ../subprojects/dpdk/lib/eal/x86/include/rte_vect.h:26,
>                  from ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:5:
> /usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h: In function ‘dir24_8_vec_lookup_x16’:
> /usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h:1947:1: error: inlining failed in call to ‘always_inline’ ‘_mm512_shuffle_epi8’: target specific option mismatch
>  1947 | _mm512_shuffle_epi8 (__m512i __A, __m512i __B)
>       | ^~~~~~~~~~~~~~~~~~~
> ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:38:26: note: called from here
>    38 |                 ip_vec = _mm512_shuffle_epi8(ip_vec, bswap32);
>       |                          ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
> /usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h:1947:1: error: inlining failed in call to ‘always_inline’ ‘_mm512_shuffle_epi8’: target specific option mismatch
>  1947 | _mm512_shuffle_epi8 (__m512i __A, __m512i __B)
>       | ^~~~~~~~~~~~~~~~~~~
> ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:38:26: note: called from here
>    38 |                 ip_vec = _mm512_shuffle_epi8(ip_vec, bswap32);
>       |                          ^~~~~~~~~

The latter issue is because you had applied the endianness change.
I replied in the other thread.


-- 
David Marchand

Re: [PATCH] fib: implement RCU rule reclamation

[Vladimir Medvedkin]

[-- Attachment #1: Type: text/plain, Size: 3791 bytes --]

Hi David, Robin,

Thanks, I'll send v2 with the fix.

сб, 28 сент. 2024 г. в 00:59, David Marchand <david.marchand@redhat.com>:

> On Fri, Sep 27, 2024 at 6:13 PM Robin Jarry <rjarry@redhat.com> wrote:
> >
> > Vladimir Medvedkin, Sep 06, 2024 at 13:09:
> > > Currently, for DIR24-8 algorithm, the tbl8 group is freed even though
> the
> > > readers might be using the tbl8 group entries. The freed tbl8 group can
> > > be reallocated quickly. As a result, lookup may be performed
> incorrectly.
> > >
> > > To address that, RCU QSBR is integrated for safe tbl8 group
> reclamation.
> > >
> > > Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
> > > ---
> >
> > > diff --git a/lib/fib/meson.build b/lib/fib/meson.build
> > > index 6795f41a0a..1895f37050 100644
> > > --- a/lib/fib/meson.build
> > > +++ b/lib/fib/meson.build
> > > @@ -11,6 +11,7 @@ endif
> > >  sources = files('rte_fib.c', 'rte_fib6.c', 'dir24_8.c', 'trie.c')
> > >  headers = files('rte_fib.h', 'rte_fib6.h')
> > >  deps += ['rib']
> > > +deps += ['rcu']
> >
> > Hi Vladimir,
> >
> > thanks a lot for working on this!
> >
> > I tested with static linking and there is a missing dependency to
> > static_rte_rcu:
> >
> > In file included from ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:6:
> > ../subprojects/dpdk/lib/fib/rte_fib.h:19:10: fatal error:
> rte_rcu_qsbr.h: No such file or directory
> >    19 | #include <rte_rcu_qsbr.h>
> >       |          ^~~~~~~~~~~~~~~~
> >
> > After adding it:
> >
> > @@ -45,7 +45,7 @@ if dpdk_conf.has('RTE_ARCH_X86_64') and binutils_ok
> >      elif cc.has_multi_arguments('-mavx512f', '-mavx512dq')
> >          dir24_8_avx512_tmp = static_library('dir24_8_avx512_tmp',
> >                  'dir24_8_avx512.c',
> > -                dependencies: static_rte_eal,
> > +                dependencies: [static_rte_eal, static_rte_rcu],
> >                  c_args: cflags + ['-mavx512f', '-mavx512dq'])
> >          objs += dir24_8_avx512_tmp.extract_objects('dir24_8_avx512.c')
> >          cflags += ['-DCC_DIR24_8_AVX512_SUPPORT']
> >
> > I get another error:
> >
> > In file included from
> /usr/lib/gcc/x86_64-redhat-linux/14/include/immintrin.h:65,
> >                  from
> /usr/lib/gcc/x86_64-redhat-linux/14/include/x86intrin.h:32,
> >                  from
> ../subprojects/dpdk/lib/eal/x86/include/rte_vect.h:26,
> >                  from ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:5:
> > /usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h: In
> function ‘dir24_8_vec_lookup_x16’:
> > /usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h:1947:1:
> error: inlining failed in call to ‘always_inline’ ‘_mm512_shuffle_epi8’:
> target specific option mismatch
> >  1947 | _mm512_shuffle_epi8 (__m512i __A, __m512i __B)
> >       | ^~~~~~~~~~~~~~~~~~~
> > ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:38:26: note: called from
> here
> >    38 |                 ip_vec = _mm512_shuffle_epi8(ip_vec, bswap32);
> >       |                          ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
> > /usr/lib/gcc/x86_64-redhat-linux/14/include/avx512bwintrin.h:1947:1:
> error: inlining failed in call to ‘always_inline’ ‘_mm512_shuffle_epi8’:
> target specific option mismatch
> >  1947 | _mm512_shuffle_epi8 (__m512i __A, __m512i __B)
> >       | ^~~~~~~~~~~~~~~~~~~
> > ../subprojects/dpdk/lib/fib/dir24_8_avx512.c:38:26: note: called from
> here
> >    38 |                 ip_vec = _mm512_shuffle_epi8(ip_vec, bswap32);
> >       |                          ^~~~~~~~~
>
> The latter issue is because you had applied the endianness change.
> I replied in the other thread.
>
>
> --
> David Marchand
>
>

-- 
Regards,
Vladimir

[-- Attachment #2: Type: text/html, Size: 5079 bytes --]

[PATCH v2 1/2] fib: implement RCU rule reclamation

[Vladimir Medvedkin]

Currently, for DIR24-8 algorithm, the tbl8 group is freed even though the
readers might be using the tbl8 group entries. The freed tbl8 group can
be reallocated quickly. As a result, lookup may be performed incorrectly.

To address that, RCU QSBR is integrated for safe tbl8 group reclamation.

Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
---
 lib/fib/dir24_8.c   | 104 +++++++++++++++++++++++++++++++++++++++-----
 lib/fib/dir24_8.h   |   9 ++++
 lib/fib/meson.build |   5 ++-
 lib/fib/rte_fib.c   |  11 +++++
 lib/fib/rte_fib.h   |  50 ++++++++++++++++++++-
 lib/fib/version.map |   7 +++
 6 files changed, 173 insertions(+), 13 deletions(-)

diff --git a/lib/fib/dir24_8.c b/lib/fib/dir24_8.c
index c739e92304..32a71c157d 100644
--- a/lib/fib/dir24_8.c
+++ b/lib/fib/dir24_8.c
@@ -14,6 +14,7 @@
 #include <rte_rib.h>
 #include <rte_fib.h>
 #include "dir24_8.h"
+#include "fib_log.h"
 
 #ifdef CC_DIR24_8_AVX512_SUPPORT
 
@@ -176,6 +177,13 @@ tbl8_alloc(struct dir24_8_tbl *dp, uint64_t nh)
 	uint8_t	*tbl8_ptr;
 
 	tbl8_idx = tbl8_get_idx(dp);
+	if ((tbl8_idx == -ENOSPC) && dp->dq != NULL) {
+		/* If there are no tbl8 groups try to reclaim one. */
+		if (rte_rcu_qsbr_dq_reclaim(dp->dq, 1,
+				NULL, NULL, NULL) == 0)
+			tbl8_idx = tbl8_get_idx(dp);
+	}
+
 	if (tbl8_idx < 0)
 		return tbl8_idx;
 	tbl8_ptr = (uint8_t *)dp->tbl8 +
@@ -189,6 +197,27 @@ tbl8_alloc(struct dir24_8_tbl *dp, uint64_t nh)
 	return tbl8_idx;
 }
 
+static void
+tbl8_cleanup_and_free(struct dir24_8_tbl *dp, uint64_t tbl8_idx)
+{
+	uint8_t *ptr = (uint8_t *)dp->tbl8 +
+		(tbl8_idx * DIR24_8_TBL8_GRP_NUM_ENT << dp->nh_sz);
+
+	memset(ptr, 0, DIR24_8_TBL8_GRP_NUM_ENT << dp->nh_sz);
+	tbl8_free_idx(dp, tbl8_idx);
+	dp->cur_tbl8s--;
+}
+
+static void
+__rcu_qsbr_free_resource(void *p, void *data, unsigned int n)
+{
+	struct dir24_8_tbl *dp = p;
+	uint64_t tbl8_idx = *(uint64_t *)data;
+	RTE_SET_USED(n);
+
+	tbl8_cleanup_and_free(dp, tbl8_idx);
+}
+
 static void
 tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 {
@@ -210,8 +239,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint8_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr8[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_2B:
 		ptr16 = &((uint16_t *)dp->tbl8)[tbl8_idx *
@@ -223,8 +250,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint16_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr16[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_4B:
 		ptr32 = &((uint32_t *)dp->tbl8)[tbl8_idx *
@@ -236,8 +261,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint32_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr32[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_8B:
 		ptr64 = &((uint64_t *)dp->tbl8)[tbl8_idx *
@@ -249,12 +272,20 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint64_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr64[i] = 0;
 		break;
 	}
-	tbl8_free_idx(dp, tbl8_idx);
-	dp->cur_tbl8s--;
+
+	if (dp->v == NULL)
+		tbl8_cleanup_and_free(dp, tbl8_idx);
+	else if (dp->rcu_mode == RTE_FIB_QSBR_MODE_SYNC) {
+		rte_rcu_qsbr_synchronize(dp->v,
+			RTE_QSBR_THRID_INVALID);
+		tbl8_cleanup_and_free(dp, tbl8_idx);
+	} else { /* RTE_FIB_QSBR_MODE_DQ */
+		if (rte_rcu_qsbr_dq_enqueue(dp->dq,
+				(void *)&tbl8_idx))
+			FIB_LOG(ERR, "Failed to push QSBR FIFO");
+	}
 }
 
 static int
@@ -569,7 +600,60 @@ dir24_8_free(void *p)
 {
 	struct dir24_8_tbl *dp = (struct dir24_8_tbl *)p;
 
+	if (dp->dq != NULL)
+		rte_rcu_qsbr_dq_delete(dp->dq);
+
 	rte_free(dp->tbl8_idxes);
 	rte_free(dp->tbl8);
 	rte_free(dp);
 }
+
+int
+dir24_8_rcu_qsbr_add(struct dir24_8_tbl *dp, struct rte_fib_rcu_config *cfg,
+	const char *name)
+{
+	struct rte_rcu_qsbr_dq_parameters params = {0};
+	char rcu_dq_name[RTE_RCU_QSBR_DQ_NAMESIZE];
+
+	if (dp == NULL || cfg == NULL) {
+		rte_errno = EINVAL;
+		return 1;
+	}
+
+	if (dp->v != NULL) {
+		rte_errno = EEXIST;
+		return 1;
+	}
+
+	if (cfg->mode == RTE_FIB_QSBR_MODE_SYNC) {
+		/* No other things to do. */
+	} else if (cfg->mode == RTE_FIB_QSBR_MODE_DQ) {
+		/* Init QSBR defer queue. */
+		snprintf(rcu_dq_name, sizeof(rcu_dq_name),
+				"FIB_RCU_%s", name);
+		params.name = rcu_dq_name;
+		params.size = cfg->dq_size;
+		if (params.size == 0)
+			params.size = RTE_FIB_RCU_DQ_RECLAIM_SZ;
+		params.trigger_reclaim_limit = cfg->reclaim_thd;
+		params.max_reclaim_size = cfg->reclaim_max;
+		if (params.max_reclaim_size == 0)
+			params.max_reclaim_size = RTE_FIB_RCU_DQ_RECLAIM_MAX;
+		params.esize = sizeof(uint64_t);
+		params.free_fn = __rcu_qsbr_free_resource;
+		params.p = dp;
+		params.v = cfg->v;
+		dp->dq = rte_rcu_qsbr_dq_create(&params);
+		if (dp->dq == NULL) {
+			FIB_LOG(ERR, "LPM defer queue creation failed");
+			return 1;
+		}
+	} else {
+		rte_errno = EINVAL;
+		return 1;
+	}
+	dp->rcu_mode = cfg->mode;
+	dp->v = cfg->v;
+
+	return 0;
+}
diff --git a/lib/fib/dir24_8.h b/lib/fib/dir24_8.h
index 7125049f15..08fd818ce4 100644
--- a/lib/fib/dir24_8.h
+++ b/lib/fib/dir24_8.h
@@ -10,6 +10,7 @@
 
 #include <rte_prefetch.h>
 #include <rte_branch_prediction.h>
+#include <rte_rcu_qsbr.h>
 
 /**
  * @file
@@ -30,6 +31,10 @@ struct dir24_8_tbl {
 	uint32_t	rsvd_tbl8s;	/**< Number of reserved tbl8s */
 	uint32_t	cur_tbl8s;	/**< Current number of tbl8s */
 	enum rte_fib_dir24_8_nh_sz	nh_sz;	/**< Size of nexthop entry */
+	/* RCU config. */
+	enum rte_fib_qsbr_mode rcu_mode;/* Blocking, defer queue. */
+	struct rte_rcu_qsbr *v;		/* RCU QSBR variable. */
+	struct rte_rcu_qsbr_dq *dq;	/* RCU QSBR defer queue. */
 	uint64_t	def_nh;		/**< Default next hop */
 	uint64_t	*tbl8;		/**< tbl8 table. */
 	uint64_t	*tbl8_idxes;	/**< bitmap containing free tbl8 idxes*/
@@ -250,4 +255,8 @@ int
 dir24_8_modify(struct rte_fib *fib, uint32_t ip, uint8_t depth,
 	uint64_t next_hop, int op);
 
+int
+dir24_8_rcu_qsbr_add(struct dir24_8_tbl *dp, struct rte_fib_rcu_config *cfg,
+	const char *name);
+
 #endif /* _DIR24_8_H_ */
diff --git a/lib/fib/meson.build b/lib/fib/meson.build
index 6795f41a0a..9b7477c756 100644
--- a/lib/fib/meson.build
+++ b/lib/fib/meson.build
@@ -11,6 +11,7 @@ endif
 sources = files('rte_fib.c', 'rte_fib6.c', 'dir24_8.c', 'trie.c')
 headers = files('rte_fib.h', 'rte_fib6.h')
 deps += ['rib']
+deps += ['rcu']
 
 # compile AVX512 version if:
 # we are building 64-bit binary AND binutils can generate proper code
@@ -45,7 +46,7 @@ if dpdk_conf.has('RTE_ARCH_X86_64') and binutils_ok
     elif cc.has_multi_arguments('-mavx512f', '-mavx512dq')
         dir24_8_avx512_tmp = static_library('dir24_8_avx512_tmp',
                 'dir24_8_avx512.c',
-                dependencies: static_rte_eal,
+                dependencies: [static_rte_eal, static_rte_rcu],
                 c_args: cflags + ['-mavx512f', '-mavx512dq'])
         objs += dir24_8_avx512_tmp.extract_objects('dir24_8_avx512.c')
         cflags += ['-DCC_DIR24_8_AVX512_SUPPORT']
@@ -54,7 +55,7 @@ if dpdk_conf.has('RTE_ARCH_X86_64') and binutils_ok
         if cc.has_argument('-mavx512bw')
             trie_avx512_tmp = static_library('trie_avx512_tmp',
                 'trie_avx512.c',
-                dependencies: static_rte_eal,
+                dependencies: [static_rte_eal, static_rte_rcu],
                 c_args: cflags + ['-mavx512f', \
                     '-mavx512dq', '-mavx512bw'])
             objs += trie_avx512_tmp.extract_objects('trie_avx512.c')
diff --git a/lib/fib/rte_fib.c b/lib/fib/rte_fib.c
index 4f9fba5a4f..730f50c1ba 100644
--- a/lib/fib/rte_fib.c
+++ b/lib/fib/rte_fib.c
@@ -338,3 +338,14 @@ rte_fib_select_lookup(struct rte_fib *fib,
 		return -EINVAL;
 	}
 }
+
+int
+rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg)
+{
+	switch (fib->type) {
+	case RTE_FIB_DIR24_8:
+		return dir24_8_rcu_qsbr_add(fib->dp, cfg, fib->name);
+	default:
+		return -ENOTSUP;
+	}
+}
diff --git a/lib/fib/rte_fib.h b/lib/fib/rte_fib.h
index d7a5aafe53..346eb7f149 100644
--- a/lib/fib/rte_fib.h
+++ b/lib/fib/rte_fib.h
@@ -16,7 +16,7 @@
  */
 
 #include <stdint.h>
-
+#include <rte_rcu_qsbr.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -28,6 +28,19 @@ struct rte_rib;
 /** Maximum depth value possible for IPv4 FIB. */
 #define RTE_FIB_MAXDEPTH	32
 
+/** @internal Default RCU defer queue entries to reclaim in one go. */
+#define RTE_FIB_RCU_DQ_RECLAIM_MAX	16
+/** @internal Default RCU defer queue size. */
+#define RTE_FIB_RCU_DQ_RECLAIM_SZ	128
+
+/** RCU reclamation modes */
+enum rte_fib_qsbr_mode {
+	/** Create defer queue for reclaim. */
+	RTE_FIB_QSBR_MODE_DQ = 0,
+	/** Use blocking mode reclaim. No defer queue created. */
+	RTE_FIB_QSBR_MODE_SYNC
+};
+
 /** Type of FIB struct */
 enum rte_fib_type {
 	RTE_FIB_DUMMY,		/**< RIB tree based FIB */
@@ -89,6 +102,22 @@ struct rte_fib_conf {
 	};
 };
 
+/** FIB RCU QSBR configuration structure. */
+struct rte_fib_rcu_config {
+	struct rte_rcu_qsbr *v;	/* RCU QSBR variable. */
+	/* Mode of RCU QSBR. RTE_FIB_QSBR_MODE_xxx
+	 * '0' for default: create defer queue for reclaim.
+	 */
+	enum rte_fib_qsbr_mode mode;
+	uint32_t dq_size;	/* RCU defer queue size.
+				 * default: RTE_FIB_RCU_DQ_RECLAIM_SZ.
+				 */
+	uint32_t reclaim_thd;	/* Threshold to trigger auto reclaim. */
+	uint32_t reclaim_max;	/* Max entries to reclaim in one go.
+				 * default: RTE_FIB_RCU_DQ_RECLAIM_MAX.
+				 */
+};
+
 /**
  * Create FIB
  *
@@ -219,6 +248,25 @@ rte_fib_get_rib(struct rte_fib *fib);
 int
 rte_fib_select_lookup(struct rte_fib *fib, enum rte_fib_lookup_type type);
 
+/**
+ * Associate RCU QSBR variable with a FIB object.
+ *
+ * @param fib
+ *   the fib object to add RCU QSBR
+ * @param cfg
+ *   RCU QSBR configuration
+ * @return
+ *   On success - 0
+ *   On error - 1 with error code set in rte_errno.
+ *   Possible rte_errno codes are:
+ *   - EINVAL - invalid pointer
+ *   - EEXIST - already added QSBR
+ *   - ENOMEM - memory allocation failure
+ *   - ENOTSUP - not supported by configured dataplane algorithm
+ */
+__rte_experimental
+int rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/fib/version.map b/lib/fib/version.map
index c6d2769611..df8f113df3 100644
--- a/lib/fib/version.map
+++ b/lib/fib/version.map
@@ -22,3 +22,10 @@ DPDK_25 {
 
 	local: *;
 };
+
+EXPERIMENTAL {
+	global:
+
+	# added in 24.11
+	rte_fib_rcu_qsbr_add;
+};
-- 
2.34.1

[PATCH v2 2/2] test/fib: add RCU functional tests

[Vladimir Medvedkin]

Add positive and negative tests for API rte_fib_rcu_qsbr_add.
Also test FIB library behavior when RCU QSBR is enabled.

Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
---
 app/test/test_fib.c | 209 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 209 insertions(+)

diff --git a/app/test/test_fib.c b/app/test/test_fib.c
index 45dccca1f6..ed5cfc1fdf 100644
--- a/app/test/test_fib.c
+++ b/app/test/test_fib.c
@@ -10,6 +10,7 @@
 #include <rte_ip.h>
 #include <rte_log.h>
 #include <rte_fib.h>
+#include <rte_malloc.h>
 
 #include "test.h"
 
@@ -21,6 +22,8 @@ static int32_t test_free_null(void);
 static int32_t test_add_del_invalid(void);
 static int32_t test_get_invalid(void);
 static int32_t test_lookup(void);
+static int32_t test_invalid_rcu(void);
+static int32_t test_fib_rcu_sync_rw(void);
 
 #define MAX_ROUTES	(1 << 16)
 #define MAX_TBL8	(1 << 15)
@@ -376,6 +379,210 @@ test_lookup(void)
 	return TEST_SUCCESS;
 }
 
+/*
+ * rte_fib_rcu_qsbr_add positive and negative tests.
+ *  - Add RCU QSBR variable to FIB
+ *  - Add another RCU QSBR variable to FIB
+ *  - Check returns
+ */
+int32_t
+test_invalid_rcu(void)
+{
+	struct rte_fib *fib = NULL;
+	struct rte_fib_conf config;
+	size_t sz;
+	struct rte_rcu_qsbr *qsv;
+	struct rte_rcu_qsbr *qsv2;
+	int32_t status;
+	struct rte_fib_rcu_config rcu_cfg = {0};
+	uint64_t def_nh = 100;
+
+	config.max_routes = MAX_ROUTES;
+	config.rib_ext_sz = 0;
+	config.default_nh = def_nh;
+	config.type = RTE_FIB_DUMMY;
+
+	fib = rte_fib_create(__func__, SOCKET_ID_ANY, &config);
+	RTE_TEST_ASSERT(fib != NULL, "Failed to create FIB\n");
+
+	/* Create RCU QSBR variable */
+	sz = rte_rcu_qsbr_get_memsize(RTE_MAX_LCORE);
+	qsv = (struct rte_rcu_qsbr *)rte_zmalloc_socket(NULL, sz,
+					RTE_CACHE_LINE_SIZE, SOCKET_ID_ANY);
+	RTE_TEST_ASSERT(qsv != NULL, "Can not allocate memory for RCU\n");
+
+	status = rte_rcu_qsbr_init(qsv, RTE_MAX_LCORE);
+	RTE_TEST_ASSERT(status == 0, "Can not initialize RCU\n");
+
+	rcu_cfg.v = qsv;
+
+	/* adding rcu to RTE_FIB_DUMMY FIB type */
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_SYNC;
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status == -ENOTSUP, "rte_fib_rcu_qsbr_add returned wrong error status\n");
+	rte_fib_free(fib);
+
+	/* Invalid QSBR mode */
+	config.type = RTE_FIB_DIR24_8;
+	config.dir24_8.nh_sz = RTE_FIB_DIR24_8_4B;
+	config.dir24_8.num_tbl8 = MAX_TBL8;
+	fib = rte_fib_create(__func__, SOCKET_ID_ANY, &config);
+	RTE_TEST_ASSERT(fib != NULL, "Failed to create FIB\n");
+	rcu_cfg.mode = 2;
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status != 0, "Failed to add RCU\n");
+
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_DQ;
+	/* Attach RCU QSBR to FIB */
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status == 0, "Can not attach RCU to FIB\n");
+
+	/* Create and attach another RCU QSBR to FIB table */
+	qsv2 = (struct rte_rcu_qsbr *)rte_zmalloc_socket(NULL, sz,
+					RTE_CACHE_LINE_SIZE, SOCKET_ID_ANY);
+	RTE_TEST_ASSERT(qsv2 != NULL, "Can not allocate memory for RCU\n");
+
+	rcu_cfg.v = qsv2;
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_SYNC;
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status != 0, "Secondary RCU was mistakenly attached\n");
+
+	rte_fib_free(fib);
+	rte_free(qsv);
+	rte_free(qsv2);
+
+	return TEST_SUCCESS;
+}
+
+static struct rte_fib *g_fib;
+static struct rte_rcu_qsbr *g_v;
+static uint32_t g_ip = RTE_IPV4(192, 0, 2, 100);
+static volatile uint8_t writer_done;
+/* Report quiescent state interval every 1024 lookups. Larger critical
+ * sections in reader will result in writer polling multiple times.
+ */
+#define QSBR_REPORTING_INTERVAL 1024
+#define WRITER_ITERATIONS	512
+
+/*
+ * Reader thread using rte_fib data structure with RCU.
+ */
+static int
+test_fib_rcu_qsbr_reader(void *arg)
+{
+	int i;
+	uint64_t next_hop_return = 0;
+
+	RTE_SET_USED(arg);
+	/* Register this thread to report quiescent state */
+	rte_rcu_qsbr_thread_register(g_v, 0);
+	rte_rcu_qsbr_thread_online(g_v, 0);
+
+	do {
+		for (i = 0; i < QSBR_REPORTING_INTERVAL; i++)
+			rte_fib_lookup_bulk(g_fib, &g_ip, &next_hop_return, 1);
+
+		/* Update quiescent state */
+		rte_rcu_qsbr_quiescent(g_v, 0);
+	} while (!writer_done);
+
+	rte_rcu_qsbr_thread_offline(g_v, 0);
+	rte_rcu_qsbr_thread_unregister(g_v, 0);
+
+	return 0;
+}
+
+/*
+ * rte_fib_rcu_qsbr_add sync mode functional test.
+ * 1 Reader and 1 writer. They cannot be in the same thread in this test.
+ *  - Create FIB which supports 1 tbl8 group at max
+ *  - Add RCU QSBR variable with sync mode to FIB
+ *  - Register a reader thread. Reader keeps looking up a specific rule.
+ *  - Writer keeps adding and deleting a specific rule with depth=28 (> 24)
+ */
+int32_t
+test_fib_rcu_sync_rw(void)
+{
+	struct rte_fib_conf config;
+	size_t sz;
+	int32_t status;
+	uint32_t i, next_hop;
+	uint8_t depth;
+	struct rte_fib_rcu_config rcu_cfg = {0};
+	uint64_t def_nh = 100;
+
+	if (rte_lcore_count() < 2) {
+		printf("Not enough cores for %s, expecting at least 2\n",
+			__func__);
+		return TEST_SKIPPED;
+	}
+
+	config.max_routes = MAX_ROUTES;
+	config.rib_ext_sz = 0;
+	config.default_nh = def_nh;
+	config.type = RTE_FIB_DIR24_8;
+	config.dir24_8.nh_sz = RTE_FIB_DIR24_8_4B;
+	config.dir24_8.num_tbl8 = 1;
+
+	g_fib = rte_fib_create(__func__, SOCKET_ID_ANY, &config);
+	RTE_TEST_ASSERT(g_fib != NULL, "Failed to create FIB\n");
+
+	/* Create RCU QSBR variable */
+	sz = rte_rcu_qsbr_get_memsize(1);
+	g_v = (struct rte_rcu_qsbr *)rte_zmalloc_socket(NULL, sz,
+				RTE_CACHE_LINE_SIZE, SOCKET_ID_ANY);
+	RTE_TEST_ASSERT(g_v != NULL, "Can not allocate memory for RCU\n");
+
+	status = rte_rcu_qsbr_init(g_v, 1);
+	RTE_TEST_ASSERT(status == 0, "Can not initialize RCU\n");
+
+	rcu_cfg.v = g_v;
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_SYNC;
+	/* Attach RCU QSBR to FIB table */
+	status = rte_fib_rcu_qsbr_add(g_fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status == 0, "Can not attach RCU to FIB\n");
+
+	writer_done = 0;
+	/* Launch reader thread */
+	rte_eal_remote_launch(test_fib_rcu_qsbr_reader, NULL,
+				rte_get_next_lcore(-1, 1, 0));
+
+	depth = 28;
+	next_hop = 1;
+	status = rte_fib_add(g_fib, g_ip, depth, next_hop);
+	if (status != 0) {
+		printf("%s: Failed to add rule\n", __func__);
+		goto error;
+	}
+
+	/* Writer update */
+	for (i = 0; i < WRITER_ITERATIONS; i++) {
+		status = rte_fib_delete(g_fib, g_ip, depth);
+		if (status != 0) {
+			printf("%s: Failed to delete rule at iteration %d\n",
+				__func__, i);
+			goto error;
+		}
+
+		status = rte_fib_add(g_fib, g_ip, depth, next_hop);
+		if (status != 0) {
+			printf("%s: Failed to add rule at iteration %d\n",
+				__func__, i);
+			goto error;
+		}
+	}
+
+error:
+	writer_done = 1;
+	/* Wait until reader exited. */
+	rte_eal_mp_wait_lcore();
+
+	rte_fib_free(g_fib);
+	rte_free(g_v);
+
+	return (status == 0) ? TEST_SUCCESS : TEST_FAILED;
+}
+
 static struct unit_test_suite fib_fast_tests = {
 	.suite_name = "fib autotest",
 	.setup = NULL,
@@ -386,6 +593,8 @@ static struct unit_test_suite fib_fast_tests = {
 	TEST_CASE(test_add_del_invalid),
 	TEST_CASE(test_get_invalid),
 	TEST_CASE(test_lookup),
+	TEST_CASE(test_invalid_rcu),
+	TEST_CASE(test_fib_rcu_sync_rw),
 	TEST_CASES_END()
 	}
 };
-- 
2.34.1

Re: [PATCH v2 1/2] fib: implement RCU rule reclamation

[Stephen Hemminger]

On Tue,  8 Oct 2024 17:55:23 +0000
Vladimir Medvedkin <vladimir.medvedkin@intel.com> wrote:

> @@ -569,7 +600,60 @@ dir24_8_free(void *p)
>  {
>  	struct dir24_8_tbl *dp = (struct dir24_8_tbl *)p;
>  
> +	if (dp->dq != NULL)
> +		rte_rcu_qsbr_dq_delete(dp->dq);
> +

Side note:
rte_rcu_qsbr_dq_delete should be changed to accept NULL as nop.
Like all the other free routines

Re: [PATCH v2 1/2] fib: implement RCU rule reclamation

[Stephen Hemminger]

On Tue,  8 Oct 2024 17:55:23 +0000
Vladimir Medvedkin <vladimir.medvedkin@intel.com> wrote:
> +	if ((tbl8_idx == -ENOSPC) && dp->dq != NULL) {

Better to either drop the parenthesis here, or put it on both conditions.

> +		/* If there are no tbl8 groups try to reclaim one. */
> +		if (rte_rcu_qsbr_dq_reclaim(dp->dq, 1,
> +				NULL, NULL, NULL) == 0)
> +			tbl8_idx = tbl8_get_idx(dp);
> +	}

Could add unlikely() to this expression.

	/* If there are no tbl8 groups try to reclaim one. */
	if (unlikely(tbl8_idx == -ENOSPC && dp->dq &&
		     !rte_rcu_qsbr_dq_reclaim(dp->dq, 1, NULL, NULL, NULL)))
	    tbl8_idx = tbl8_get_idx(dp);


> +static void
> +__rcu_qsbr_free_resource(void *p, void *data, unsigned int n)
> +{
> +	struct dir24_8_tbl *dp = p;
> +	uint64_t tbl8_idx = *(uint64_t *)data;
> +	RTE_SET_USED(n);
> +
> +	tbl8_cleanup_and_free(dp, tbl8_idx);
> +}

My preference (not a requirement) is to use __rte_unused attribute
instead of RTE_SET_USED

> +	if (dp->v == NULL)
> +		tbl8_cleanup_and_free(dp, tbl8_idx);
> +	else if (dp->rcu_mode == RTE_FIB_QSBR_MODE_SYNC) {
> +		rte_rcu_qsbr_synchronize(dp->v,
> +			RTE_QSBR_THRID_INVALID);
> +		tbl8_cleanup_and_free(dp, tbl8_idx);
> +	} else { /* RTE_FIB_QSBR_MODE_DQ */
> +		if (rte_rcu_qsbr_dq_enqueue(dp->dq,
> +				(void *)&tbl8_idx))

Minor nit: cast to void * is not necessary in C (only in C++).
And can fit on one line; max line length now for DPDK is 100 characters.

Overall, looks good.

Acked-by: Stephen Hemminger <stephen@networkplumber.org>

RE: [PATCH v2 1/2] fib: implement RCU rule reclamation

[Doug Foster]

The check for NULL is not necessary before calling rte_rcu_qsbr_dq_delete. Similar to other free routines, an error will not occur when the dq pointer is NULL.
However, it will give a debug log statement to indicate an invalid parameter and return 0 to indicate success.

-----Original Message-----
From: Stephen Hemminger <stephen@networkplumber.org>
Sent: Tuesday, October 8, 2024 1:18 PM
To: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
Cc: dev@dpdk.org; rjarry@redhat.com; Ruifeng Wang <Ruifeng.Wang@arm.com>; Honnappa Nagarahalli <Honnappa.Nagarahalli@arm.com>; david.marchand@redhat.com
Subject: Re: [PATCH v2 1/2] fib: implement RCU rule reclamation

On Tue,  8 Oct 2024 17:55:23 +0000
Vladimir Medvedkin <vladimir.medvedkin@intel.com> wrote:

> @@ -569,7 +600,60 @@ dir24_8_free(void *p)  {
>       struct dir24_8_tbl *dp = (struct dir24_8_tbl *)p;
>
> +     if (dp->dq != NULL)
> +             rte_rcu_qsbr_dq_delete(dp->dq);
> +

Side note:
rte_rcu_qsbr_dq_delete should be changed to accept NULL as nop.
Like all the other free routines
IMPORTANT NOTICE: The contents of this email and any attachments are confidential and may also be privileged. If you are not the intended recipient, please notify the sender immediately and do not disclose the contents to any other person, use it for any purpose, or store or copy the information in any medium. Thank you.

Re: [PATCH v2 1/2] fib: implement RCU rule reclamation

[Medvedkin, Vladimir]

Hi Stephen,

Thanks for the review, I'll address your comments in v3


On 08/10/2024 19:28, Stephen Hemminger wrote:
> On Tue,  8 Oct 2024 17:55:23 +0000
> Vladimir Medvedkin <vladimir.medvedkin@intel.com> wrote:
>> +	if ((tbl8_idx == -ENOSPC) && dp->dq != NULL) {
> Better to either drop the parenthesis here, or put it on both conditions.
>
>> +		/* If there are no tbl8 groups try to reclaim one. */
>> +		if (rte_rcu_qsbr_dq_reclaim(dp->dq, 1,
>> +				NULL, NULL, NULL) == 0)
>> +			tbl8_idx = tbl8_get_idx(dp);
>> +	}
> Could add unlikely() to this expression.
>
> 	/* If there are no tbl8 groups try to reclaim one. */
> 	if (unlikely(tbl8_idx == -ENOSPC && dp->dq &&
> 		     !rte_rcu_qsbr_dq_reclaim(dp->dq, 1, NULL, NULL, NULL)))
> 	    tbl8_idx = tbl8_get_idx(dp);
>
>
>> +static void
>> +__rcu_qsbr_free_resource(void *p, void *data, unsigned int n)
>> +{
>> +	struct dir24_8_tbl *dp = p;
>> +	uint64_t tbl8_idx = *(uint64_t *)data;
>> +	RTE_SET_USED(n);
>> +
>> +	tbl8_cleanup_and_free(dp, tbl8_idx);
>> +}
> My preference (not a requirement) is to use __rte_unused attribute
> instead of RTE_SET_USED
>
>> +	if (dp->v == NULL)
>> +		tbl8_cleanup_and_free(dp, tbl8_idx);
>> +	else if (dp->rcu_mode == RTE_FIB_QSBR_MODE_SYNC) {
>> +		rte_rcu_qsbr_synchronize(dp->v,
>> +			RTE_QSBR_THRID_INVALID);
>> +		tbl8_cleanup_and_free(dp, tbl8_idx);
>> +	} else { /* RTE_FIB_QSBR_MODE_DQ */
>> +		if (rte_rcu_qsbr_dq_enqueue(dp->dq,
>> +				(void *)&tbl8_idx))
> Minor nit: cast to void * is not necessary in C (only in C++).
> And can fit on one line; max line length now for DPDK is 100 characters.
>
> Overall, looks good.
>
> Acked-by: Stephen Hemminger <stephen@networkplumber.org>

-- 
Regards,
Vladimir

[PATCH v3 1/2] fib: implement RCU rule reclamation

[Vladimir Medvedkin]

Currently, for DIR24-8 algorithm, the tbl8 group is freed even though the
readers might be using the tbl8 group entries. The freed tbl8 group can
be reallocated quickly. As a result, lookup may be performed incorrectly.

To address that, RCU QSBR is integrated for safe tbl8 group reclamation.

Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
Acked-by: Stephen Hemminger <stephen@networkplumber.org>
---
 lib/fib/dir24_8.c   | 98 ++++++++++++++++++++++++++++++++++++++++-----
 lib/fib/dir24_8.h   |  9 +++++
 lib/fib/meson.build |  5 ++-
 lib/fib/rte_fib.c   | 11 +++++
 lib/fib/rte_fib.h   | 50 ++++++++++++++++++++++-
 lib/fib/version.map |  7 ++++
 6 files changed, 167 insertions(+), 13 deletions(-)

diff --git a/lib/fib/dir24_8.c b/lib/fib/dir24_8.c
index c739e92304..77b8fbc0db 100644
--- a/lib/fib/dir24_8.c
+++ b/lib/fib/dir24_8.c
@@ -14,6 +14,7 @@
 #include <rte_rib.h>
 #include <rte_fib.h>
 #include "dir24_8.h"
+#include "fib_log.h"
 
 #ifdef CC_DIR24_8_AVX512_SUPPORT
 
@@ -176,6 +177,12 @@ tbl8_alloc(struct dir24_8_tbl *dp, uint64_t nh)
 	uint8_t	*tbl8_ptr;
 
 	tbl8_idx = tbl8_get_idx(dp);
+
+	/* If there are no tbl8 groups try to reclaim one. */
+	if (unlikely(tbl8_idx == -ENOSPC && dp->dq &&
+		     !rte_rcu_qsbr_dq_reclaim(dp->dq, 1, NULL, NULL, NULL)))
+		tbl8_idx = tbl8_get_idx(dp);
+
 	if (tbl8_idx < 0)
 		return tbl8_idx;
 	tbl8_ptr = (uint8_t *)dp->tbl8 +
@@ -189,6 +196,26 @@ tbl8_alloc(struct dir24_8_tbl *dp, uint64_t nh)
 	return tbl8_idx;
 }
 
+static void
+tbl8_cleanup_and_free(struct dir24_8_tbl *dp, uint64_t tbl8_idx)
+{
+	uint8_t *ptr = (uint8_t *)dp->tbl8 +
+		(tbl8_idx * DIR24_8_TBL8_GRP_NUM_ENT << dp->nh_sz);
+
+	memset(ptr, 0, DIR24_8_TBL8_GRP_NUM_ENT << dp->nh_sz);
+	tbl8_free_idx(dp, tbl8_idx);
+	dp->cur_tbl8s--;
+}
+
+static void
+__rcu_qsbr_free_resource(void *p, void *data, unsigned int n __rte_unused)
+{
+	struct dir24_8_tbl *dp = p;
+	uint64_t tbl8_idx = *(uint64_t *)data;
+
+	tbl8_cleanup_and_free(dp, tbl8_idx);
+}
+
 static void
 tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 {
@@ -210,8 +237,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint8_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr8[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_2B:
 		ptr16 = &((uint16_t *)dp->tbl8)[tbl8_idx *
@@ -223,8 +248,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint16_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr16[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_4B:
 		ptr32 = &((uint32_t *)dp->tbl8)[tbl8_idx *
@@ -236,8 +259,6 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint32_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr32[i] = 0;
 		break;
 	case RTE_FIB_DIR24_8_8B:
 		ptr64 = &((uint64_t *)dp->tbl8)[tbl8_idx *
@@ -249,12 +270,18 @@ tbl8_recycle(struct dir24_8_tbl *dp, uint32_t ip, uint64_t tbl8_idx)
 		}
 		((uint64_t *)dp->tbl24)[ip >> 8] =
 			nh & ~DIR24_8_EXT_ENT;
-		for (i = 0; i < DIR24_8_TBL8_GRP_NUM_ENT; i++)
-			ptr64[i] = 0;
 		break;
 	}
-	tbl8_free_idx(dp, tbl8_idx);
-	dp->cur_tbl8s--;
+
+	if (dp->v == NULL)
+		tbl8_cleanup_and_free(dp, tbl8_idx);
+	else if (dp->rcu_mode == RTE_FIB_QSBR_MODE_SYNC) {
+		rte_rcu_qsbr_synchronize(dp->v, RTE_QSBR_THRID_INVALID);
+		tbl8_cleanup_and_free(dp, tbl8_idx);
+	} else { /* RTE_FIB_QSBR_MODE_DQ */
+		if (rte_rcu_qsbr_dq_enqueue(dp->dq, &tbl8_idx))
+			FIB_LOG(ERR, "Failed to push QSBR FIFO");
+	}
 }
 
 static int
@@ -569,7 +596,58 @@ dir24_8_free(void *p)
 {
 	struct dir24_8_tbl *dp = (struct dir24_8_tbl *)p;
 
+	rte_rcu_qsbr_dq_delete(dp->dq);
 	rte_free(dp->tbl8_idxes);
 	rte_free(dp->tbl8);
 	rte_free(dp);
 }
+
+int
+dir24_8_rcu_qsbr_add(struct dir24_8_tbl *dp, struct rte_fib_rcu_config *cfg,
+	const char *name)
+{
+	struct rte_rcu_qsbr_dq_parameters params = {0};
+	char rcu_dq_name[RTE_RCU_QSBR_DQ_NAMESIZE];
+
+	if (dp == NULL || cfg == NULL) {
+		rte_errno = EINVAL;
+		return 1;
+	}
+
+	if (dp->v != NULL) {
+		rte_errno = EEXIST;
+		return 1;
+	}
+
+	if (cfg->mode == RTE_FIB_QSBR_MODE_SYNC) {
+		/* No other things to do. */
+	} else if (cfg->mode == RTE_FIB_QSBR_MODE_DQ) {
+		/* Init QSBR defer queue. */
+		snprintf(rcu_dq_name, sizeof(rcu_dq_name),
+				"FIB_RCU_%s", name);
+		params.name = rcu_dq_name;
+		params.size = cfg->dq_size;
+		if (params.size == 0)
+			params.size = RTE_FIB_RCU_DQ_RECLAIM_SZ;
+		params.trigger_reclaim_limit = cfg->reclaim_thd;
+		params.max_reclaim_size = cfg->reclaim_max;
+		if (params.max_reclaim_size == 0)
+			params.max_reclaim_size = RTE_FIB_RCU_DQ_RECLAIM_MAX;
+		params.esize = sizeof(uint64_t);
+		params.free_fn = __rcu_qsbr_free_resource;
+		params.p = dp;
+		params.v = cfg->v;
+		dp->dq = rte_rcu_qsbr_dq_create(&params);
+		if (dp->dq == NULL) {
+			FIB_LOG(ERR, "LPM defer queue creation failed");
+			return 1;
+		}
+	} else {
+		rte_errno = EINVAL;
+		return 1;
+	}
+	dp->rcu_mode = cfg->mode;
+	dp->v = cfg->v;
+
+	return 0;
+}
diff --git a/lib/fib/dir24_8.h b/lib/fib/dir24_8.h
index 7125049f15..08fd818ce4 100644
--- a/lib/fib/dir24_8.h
+++ b/lib/fib/dir24_8.h
@@ -10,6 +10,7 @@
 
 #include <rte_prefetch.h>
 #include <rte_branch_prediction.h>
+#include <rte_rcu_qsbr.h>
 
 /**
  * @file
@@ -30,6 +31,10 @@ struct dir24_8_tbl {
 	uint32_t	rsvd_tbl8s;	/**< Number of reserved tbl8s */
 	uint32_t	cur_tbl8s;	/**< Current number of tbl8s */
 	enum rte_fib_dir24_8_nh_sz	nh_sz;	/**< Size of nexthop entry */
+	/* RCU config. */
+	enum rte_fib_qsbr_mode rcu_mode;/* Blocking, defer queue. */
+	struct rte_rcu_qsbr *v;		/* RCU QSBR variable. */
+	struct rte_rcu_qsbr_dq *dq;	/* RCU QSBR defer queue. */
 	uint64_t	def_nh;		/**< Default next hop */
 	uint64_t	*tbl8;		/**< tbl8 table. */
 	uint64_t	*tbl8_idxes;	/**< bitmap containing free tbl8 idxes*/
@@ -250,4 +255,8 @@ int
 dir24_8_modify(struct rte_fib *fib, uint32_t ip, uint8_t depth,
 	uint64_t next_hop, int op);
 
+int
+dir24_8_rcu_qsbr_add(struct dir24_8_tbl *dp, struct rte_fib_rcu_config *cfg,
+	const char *name);
+
 #endif /* _DIR24_8_H_ */
diff --git a/lib/fib/meson.build b/lib/fib/meson.build
index 6795f41a0a..9b7477c756 100644
--- a/lib/fib/meson.build
+++ b/lib/fib/meson.build
@@ -11,6 +11,7 @@ endif
 sources = files('rte_fib.c', 'rte_fib6.c', 'dir24_8.c', 'trie.c')
 headers = files('rte_fib.h', 'rte_fib6.h')
 deps += ['rib']
+deps += ['rcu']
 
 # compile AVX512 version if:
 # we are building 64-bit binary AND binutils can generate proper code
@@ -45,7 +46,7 @@ if dpdk_conf.has('RTE_ARCH_X86_64') and binutils_ok
     elif cc.has_multi_arguments('-mavx512f', '-mavx512dq')
         dir24_8_avx512_tmp = static_library('dir24_8_avx512_tmp',
                 'dir24_8_avx512.c',
-                dependencies: static_rte_eal,
+                dependencies: [static_rte_eal, static_rte_rcu],
                 c_args: cflags + ['-mavx512f', '-mavx512dq'])
         objs += dir24_8_avx512_tmp.extract_objects('dir24_8_avx512.c')
         cflags += ['-DCC_DIR24_8_AVX512_SUPPORT']
@@ -54,7 +55,7 @@ if dpdk_conf.has('RTE_ARCH_X86_64') and binutils_ok
         if cc.has_argument('-mavx512bw')
             trie_avx512_tmp = static_library('trie_avx512_tmp',
                 'trie_avx512.c',
-                dependencies: static_rte_eal,
+                dependencies: [static_rte_eal, static_rte_rcu],
                 c_args: cflags + ['-mavx512f', \
                     '-mavx512dq', '-mavx512bw'])
             objs += trie_avx512_tmp.extract_objects('trie_avx512.c')
diff --git a/lib/fib/rte_fib.c b/lib/fib/rte_fib.c
index 4f9fba5a4f..730f50c1ba 100644
--- a/lib/fib/rte_fib.c
+++ b/lib/fib/rte_fib.c
@@ -338,3 +338,14 @@ rte_fib_select_lookup(struct rte_fib *fib,
 		return -EINVAL;
 	}
 }
+
+int
+rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg)
+{
+	switch (fib->type) {
+	case RTE_FIB_DIR24_8:
+		return dir24_8_rcu_qsbr_add(fib->dp, cfg, fib->name);
+	default:
+		return -ENOTSUP;
+	}
+}
diff --git a/lib/fib/rte_fib.h b/lib/fib/rte_fib.h
index d7a5aafe53..346eb7f149 100644
--- a/lib/fib/rte_fib.h
+++ b/lib/fib/rte_fib.h
@@ -16,7 +16,7 @@
  */
 
 #include <stdint.h>
-
+#include <rte_rcu_qsbr.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -28,6 +28,19 @@ struct rte_rib;
 /** Maximum depth value possible for IPv4 FIB. */
 #define RTE_FIB_MAXDEPTH	32
 
+/** @internal Default RCU defer queue entries to reclaim in one go. */
+#define RTE_FIB_RCU_DQ_RECLAIM_MAX	16
+/** @internal Default RCU defer queue size. */
+#define RTE_FIB_RCU_DQ_RECLAIM_SZ	128
+
+/** RCU reclamation modes */
+enum rte_fib_qsbr_mode {
+	/** Create defer queue for reclaim. */
+	RTE_FIB_QSBR_MODE_DQ = 0,
+	/** Use blocking mode reclaim. No defer queue created. */
+	RTE_FIB_QSBR_MODE_SYNC
+};
+
 /** Type of FIB struct */
 enum rte_fib_type {
 	RTE_FIB_DUMMY,		/**< RIB tree based FIB */
@@ -89,6 +102,22 @@ struct rte_fib_conf {
 	};
 };
 
+/** FIB RCU QSBR configuration structure. */
+struct rte_fib_rcu_config {
+	struct rte_rcu_qsbr *v;	/* RCU QSBR variable. */
+	/* Mode of RCU QSBR. RTE_FIB_QSBR_MODE_xxx
+	 * '0' for default: create defer queue for reclaim.
+	 */
+	enum rte_fib_qsbr_mode mode;
+	uint32_t dq_size;	/* RCU defer queue size.
+				 * default: RTE_FIB_RCU_DQ_RECLAIM_SZ.
+				 */
+	uint32_t reclaim_thd;	/* Threshold to trigger auto reclaim. */
+	uint32_t reclaim_max;	/* Max entries to reclaim in one go.
+				 * default: RTE_FIB_RCU_DQ_RECLAIM_MAX.
+				 */
+};
+
 /**
  * Create FIB
  *
@@ -219,6 +248,25 @@ rte_fib_get_rib(struct rte_fib *fib);
 int
 rte_fib_select_lookup(struct rte_fib *fib, enum rte_fib_lookup_type type);
 
+/**
+ * Associate RCU QSBR variable with a FIB object.
+ *
+ * @param fib
+ *   the fib object to add RCU QSBR
+ * @param cfg
+ *   RCU QSBR configuration
+ * @return
+ *   On success - 0
+ *   On error - 1 with error code set in rte_errno.
+ *   Possible rte_errno codes are:
+ *   - EINVAL - invalid pointer
+ *   - EEXIST - already added QSBR
+ *   - ENOMEM - memory allocation failure
+ *   - ENOTSUP - not supported by configured dataplane algorithm
+ */
+__rte_experimental
+int rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/fib/version.map b/lib/fib/version.map
index c6d2769611..df8f113df3 100644
--- a/lib/fib/version.map
+++ b/lib/fib/version.map
@@ -22,3 +22,10 @@ DPDK_25 {
 
 	local: *;
 };
+
+EXPERIMENTAL {
+	global:
+
+	# added in 24.11
+	rte_fib_rcu_qsbr_add;
+};
-- 
2.34.1

[PATCH v3 2/2] test/fib: add RCU functional tests

[Vladimir Medvedkin]

Add positive and negative tests for API rte_fib_rcu_qsbr_add.
Also test FIB library behavior when RCU QSBR is enabled.

Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
---
 app/test/test_fib.c | 209 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 209 insertions(+)

diff --git a/app/test/test_fib.c b/app/test/test_fib.c
index 45dccca1f6..ed5cfc1fdf 100644
--- a/app/test/test_fib.c
+++ b/app/test/test_fib.c
@@ -10,6 +10,7 @@
 #include <rte_ip.h>
 #include <rte_log.h>
 #include <rte_fib.h>
+#include <rte_malloc.h>
 
 #include "test.h"
 
@@ -21,6 +22,8 @@ static int32_t test_free_null(void);
 static int32_t test_add_del_invalid(void);
 static int32_t test_get_invalid(void);
 static int32_t test_lookup(void);
+static int32_t test_invalid_rcu(void);
+static int32_t test_fib_rcu_sync_rw(void);
 
 #define MAX_ROUTES	(1 << 16)
 #define MAX_TBL8	(1 << 15)
@@ -376,6 +379,210 @@ test_lookup(void)
 	return TEST_SUCCESS;
 }
 
+/*
+ * rte_fib_rcu_qsbr_add positive and negative tests.
+ *  - Add RCU QSBR variable to FIB
+ *  - Add another RCU QSBR variable to FIB
+ *  - Check returns
+ */
+int32_t
+test_invalid_rcu(void)
+{
+	struct rte_fib *fib = NULL;
+	struct rte_fib_conf config;
+	size_t sz;
+	struct rte_rcu_qsbr *qsv;
+	struct rte_rcu_qsbr *qsv2;
+	int32_t status;
+	struct rte_fib_rcu_config rcu_cfg = {0};
+	uint64_t def_nh = 100;
+
+	config.max_routes = MAX_ROUTES;
+	config.rib_ext_sz = 0;
+	config.default_nh = def_nh;
+	config.type = RTE_FIB_DUMMY;
+
+	fib = rte_fib_create(__func__, SOCKET_ID_ANY, &config);
+	RTE_TEST_ASSERT(fib != NULL, "Failed to create FIB\n");
+
+	/* Create RCU QSBR variable */
+	sz = rte_rcu_qsbr_get_memsize(RTE_MAX_LCORE);
+	qsv = (struct rte_rcu_qsbr *)rte_zmalloc_socket(NULL, sz,
+					RTE_CACHE_LINE_SIZE, SOCKET_ID_ANY);
+	RTE_TEST_ASSERT(qsv != NULL, "Can not allocate memory for RCU\n");
+
+	status = rte_rcu_qsbr_init(qsv, RTE_MAX_LCORE);
+	RTE_TEST_ASSERT(status == 0, "Can not initialize RCU\n");
+
+	rcu_cfg.v = qsv;
+
+	/* adding rcu to RTE_FIB_DUMMY FIB type */
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_SYNC;
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status == -ENOTSUP, "rte_fib_rcu_qsbr_add returned wrong error status\n");
+	rte_fib_free(fib);
+
+	/* Invalid QSBR mode */
+	config.type = RTE_FIB_DIR24_8;
+	config.dir24_8.nh_sz = RTE_FIB_DIR24_8_4B;
+	config.dir24_8.num_tbl8 = MAX_TBL8;
+	fib = rte_fib_create(__func__, SOCKET_ID_ANY, &config);
+	RTE_TEST_ASSERT(fib != NULL, "Failed to create FIB\n");
+	rcu_cfg.mode = 2;
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status != 0, "Failed to add RCU\n");
+
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_DQ;
+	/* Attach RCU QSBR to FIB */
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status == 0, "Can not attach RCU to FIB\n");
+
+	/* Create and attach another RCU QSBR to FIB table */
+	qsv2 = (struct rte_rcu_qsbr *)rte_zmalloc_socket(NULL, sz,
+					RTE_CACHE_LINE_SIZE, SOCKET_ID_ANY);
+	RTE_TEST_ASSERT(qsv2 != NULL, "Can not allocate memory for RCU\n");
+
+	rcu_cfg.v = qsv2;
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_SYNC;
+	status = rte_fib_rcu_qsbr_add(fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status != 0, "Secondary RCU was mistakenly attached\n");
+
+	rte_fib_free(fib);
+	rte_free(qsv);
+	rte_free(qsv2);
+
+	return TEST_SUCCESS;
+}
+
+static struct rte_fib *g_fib;
+static struct rte_rcu_qsbr *g_v;
+static uint32_t g_ip = RTE_IPV4(192, 0, 2, 100);
+static volatile uint8_t writer_done;
+/* Report quiescent state interval every 1024 lookups. Larger critical
+ * sections in reader will result in writer polling multiple times.
+ */
+#define QSBR_REPORTING_INTERVAL 1024
+#define WRITER_ITERATIONS	512
+
+/*
+ * Reader thread using rte_fib data structure with RCU.
+ */
+static int
+test_fib_rcu_qsbr_reader(void *arg)
+{
+	int i;
+	uint64_t next_hop_return = 0;
+
+	RTE_SET_USED(arg);
+	/* Register this thread to report quiescent state */
+	rte_rcu_qsbr_thread_register(g_v, 0);
+	rte_rcu_qsbr_thread_online(g_v, 0);
+
+	do {
+		for (i = 0; i < QSBR_REPORTING_INTERVAL; i++)
+			rte_fib_lookup_bulk(g_fib, &g_ip, &next_hop_return, 1);
+
+		/* Update quiescent state */
+		rte_rcu_qsbr_quiescent(g_v, 0);
+	} while (!writer_done);
+
+	rte_rcu_qsbr_thread_offline(g_v, 0);
+	rte_rcu_qsbr_thread_unregister(g_v, 0);
+
+	return 0;
+}
+
+/*
+ * rte_fib_rcu_qsbr_add sync mode functional test.
+ * 1 Reader and 1 writer. They cannot be in the same thread in this test.
+ *  - Create FIB which supports 1 tbl8 group at max
+ *  - Add RCU QSBR variable with sync mode to FIB
+ *  - Register a reader thread. Reader keeps looking up a specific rule.
+ *  - Writer keeps adding and deleting a specific rule with depth=28 (> 24)
+ */
+int32_t
+test_fib_rcu_sync_rw(void)
+{
+	struct rte_fib_conf config;
+	size_t sz;
+	int32_t status;
+	uint32_t i, next_hop;
+	uint8_t depth;
+	struct rte_fib_rcu_config rcu_cfg = {0};
+	uint64_t def_nh = 100;
+
+	if (rte_lcore_count() < 2) {
+		printf("Not enough cores for %s, expecting at least 2\n",
+			__func__);
+		return TEST_SKIPPED;
+	}
+
+	config.max_routes = MAX_ROUTES;
+	config.rib_ext_sz = 0;
+	config.default_nh = def_nh;
+	config.type = RTE_FIB_DIR24_8;
+	config.dir24_8.nh_sz = RTE_FIB_DIR24_8_4B;
+	config.dir24_8.num_tbl8 = 1;
+
+	g_fib = rte_fib_create(__func__, SOCKET_ID_ANY, &config);
+	RTE_TEST_ASSERT(g_fib != NULL, "Failed to create FIB\n");
+
+	/* Create RCU QSBR variable */
+	sz = rte_rcu_qsbr_get_memsize(1);
+	g_v = (struct rte_rcu_qsbr *)rte_zmalloc_socket(NULL, sz,
+				RTE_CACHE_LINE_SIZE, SOCKET_ID_ANY);
+	RTE_TEST_ASSERT(g_v != NULL, "Can not allocate memory for RCU\n");
+
+	status = rte_rcu_qsbr_init(g_v, 1);
+	RTE_TEST_ASSERT(status == 0, "Can not initialize RCU\n");
+
+	rcu_cfg.v = g_v;
+	rcu_cfg.mode = RTE_FIB_QSBR_MODE_SYNC;
+	/* Attach RCU QSBR to FIB table */
+	status = rte_fib_rcu_qsbr_add(g_fib, &rcu_cfg);
+	RTE_TEST_ASSERT(status == 0, "Can not attach RCU to FIB\n");
+
+	writer_done = 0;
+	/* Launch reader thread */
+	rte_eal_remote_launch(test_fib_rcu_qsbr_reader, NULL,
+				rte_get_next_lcore(-1, 1, 0));
+
+	depth = 28;
+	next_hop = 1;
+	status = rte_fib_add(g_fib, g_ip, depth, next_hop);
+	if (status != 0) {
+		printf("%s: Failed to add rule\n", __func__);
+		goto error;
+	}
+
+	/* Writer update */
+	for (i = 0; i < WRITER_ITERATIONS; i++) {
+		status = rte_fib_delete(g_fib, g_ip, depth);
+		if (status != 0) {
+			printf("%s: Failed to delete rule at iteration %d\n",
+				__func__, i);
+			goto error;
+		}
+
+		status = rte_fib_add(g_fib, g_ip, depth, next_hop);
+		if (status != 0) {
+			printf("%s: Failed to add rule at iteration %d\n",
+				__func__, i);
+			goto error;
+		}
+	}
+
+error:
+	writer_done = 1;
+	/* Wait until reader exited. */
+	rte_eal_mp_wait_lcore();
+
+	rte_fib_free(g_fib);
+	rte_free(g_v);
+
+	return (status == 0) ? TEST_SUCCESS : TEST_FAILED;
+}
+
 static struct unit_test_suite fib_fast_tests = {
 	.suite_name = "fib autotest",
 	.setup = NULL,
@@ -386,6 +593,8 @@ static struct unit_test_suite fib_fast_tests = {
 	TEST_CASE(test_add_del_invalid),
 	TEST_CASE(test_get_invalid),
 	TEST_CASE(test_lookup),
+	TEST_CASE(test_invalid_rcu),
+	TEST_CASE(test_fib_rcu_sync_rw),
 	TEST_CASES_END()
 	}
 };
-- 
2.34.1

Re: [PATCH v3 1/2] fib: implement RCU rule reclamation

[David Marchand]

On Thu, Oct 10, 2024 at 1:27 PM Vladimir Medvedkin
<vladimir.medvedkin@intel.com> wrote:
>
> Currently, for DIR24-8 algorithm, the tbl8 group is freed even though the
> readers might be using the tbl8 group entries. The freed tbl8 group can
> be reallocated quickly. As a result, lookup may be performed incorrectly.
>
> To address that, RCU QSBR is integrated for safe tbl8 group reclamation.
>
> Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
> Acked-by: Stephen Hemminger <stephen@networkplumber.org>

We got one false positive in bitops unit test, and one strange failure on ARM.

Recheck-request: iol-unit-amd64-testing, iol-unit-arm64-testing


-- 
David Marchand

Re: [PATCH v3 1/2] fib: implement RCU rule reclamation

[David Marchand]

On Thu, Oct 10, 2024 at 1:27 PM Vladimir Medvedkin
<vladimir.medvedkin@intel.com> wrote:
> diff --git a/lib/fib/rte_fib.c b/lib/fib/rte_fib.c
> index 4f9fba5a4f..730f50c1ba 100644
> --- a/lib/fib/rte_fib.c
> +++ b/lib/fib/rte_fib.c
> @@ -338,3 +338,14 @@ rte_fib_select_lookup(struct rte_fib *fib,
>                 return -EINVAL;
>         }
>  }
> +
> +int
> +rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg)
> +{
> +       switch (fib->type) {
> +       case RTE_FIB_DIR24_8:
> +               return dir24_8_rcu_qsbr_add(fib->dp, cfg, fib->name);
> +       default:
> +               return -ENOTSUP;

This does not align with the documented API.
Please send a fix.


> +       }
> +}
> diff --git a/lib/fib/rte_fib.h b/lib/fib/rte_fib.h
> index d7a5aafe53..346eb7f149 100644

[snip]

>  /**
>   * Create FIB
>   *
> @@ -219,6 +248,25 @@ rte_fib_get_rib(struct rte_fib *fib);
>  int
>  rte_fib_select_lookup(struct rte_fib *fib, enum rte_fib_lookup_type type);
>
> +/**
> + * Associate RCU QSBR variable with a FIB object.
> + *
> + * @param fib
> + *   the fib object to add RCU QSBR
> + * @param cfg
> + *   RCU QSBR configuration
> + * @return
> + *   On success - 0
> + *   On error - 1 with error code set in rte_errno.
> + *   Possible rte_errno codes are:
> + *   - EINVAL - invalid pointer
> + *   - EEXIST - already added QSBR
> + *   - ENOMEM - memory allocation failure
> + *   - ENOTSUP - not supported by configured dataplane algorithm

In general, the fib API returns a negative integer in general.

I'll merge this patch as is for rc1 but I would prefer to have
something consistent for rc2.
Can you send a followup patch?


> + */
> +__rte_experimental
> +int rte_fib_rcu_qsbr_add(struct rte_fib *fib, struct rte_fib_rcu_config *cfg);
> +
>  #ifdef __cplusplus
>  }
>  #endif


Thanks.

-- 
David Marchand

Re: [PATCH v3 1/2] fib: implement RCU rule reclamation

[David Marchand]

On Thu, Oct 10, 2024 at 1:27 PM Vladimir Medvedkin
<vladimir.medvedkin@intel.com> wrote:
>
> Currently, for DIR24-8 algorithm, the tbl8 group is freed even though the
> readers might be using the tbl8 group entries. The freed tbl8 group can
> be reallocated quickly. As a result, lookup may be performed incorrectly.
>
> To address that, RCU QSBR is integrated for safe tbl8 group reclamation.
>
> Signed-off-by: Vladimir Medvedkin <vladimir.medvedkin@intel.com>
> Acked-by: Stephen Hemminger <stephen@networkplumber.org>

Applied (even though I sent comments), thanks.


-- 
David Marchand