[dpdk-dev] [RFC] hash/lpm: return NULL if the object exists

[Olivier Matz <olivier.matz@6wind.com>]

Seen by trying to fix the func_reentrancy autotest. The test
was doing the following on several cores in parallel:

  name = "common_name";
  do several times {
      obj = allocate_an_object(name)   // obj = ring, mempool, hash, lpm, ...
      if (obj == NULL && lookup(name) == NULL)
          return TEST_FAIL;
  }

Issues:

1/ rings, mempools, hashs API are not coherent
   rings and mempool return NULL if the object does not exist
   hash and lpm return an object that was allocated allocated if
   it already was allocated

2/ The hash/lpm API looks dangerous: when an object is returned,
   the user does not know if it should be freed or not (no refcnt)

3/ There are some possible race conditions in cuckoo_hash as the
   lock is not held in rte_hash_create(). We could find some cases
   where NULL is returned when the object already exists (ex: when
   rte_ring_create() fails).

This patch tries to rationalize the APIs of lpm and hash.

Signed-off-by: Olivier Matz <olivier.matz@6wind.com>
---
 app/test/test_func_reentrancy.c   | 31 +++++++++++++++++++++----------
 app/test/test_lpm6.c              |  2 +-
 lib/librte_hash/rte_cuckoo_hash.c |  2 +-
 lib/librte_hash/rte_fbk_hash.c    |  4 +++-
 lib/librte_lpm/rte_lpm.c          |  8 ++++++--
 lib/librte_lpm/rte_lpm6.c         |  4 +++-
 6 files changed, 35 insertions(+), 16 deletions(-)

diff --git a/app/test/test_func_reentrancy.c b/app/test/test_func_reentrancy.c
index 5d09296..300a3bc 100644
--- a/app/test/test_func_reentrancy.c
+++ b/app/test/test_func_reentrancy.c
@@ -83,6 +83,7 @@ typedef void (*case_clean_t)(unsigned lcore_id);
 
 #define MAX_LCORES	RTE_MAX_MEMZONE / (MAX_ITER_TIMES * 4U)
 
+static rte_atomic32_t obj_count = RTE_ATOMIC32_INIT(0);
 static rte_atomic32_t synchro = RTE_ATOMIC32_INIT(0);
 
 #define WAIT_SYNCHRO_FOR_SLAVES()   do{ \
@@ -100,6 +101,7 @@ test_eal_init_once(__attribute__((unused)) void *arg)
 
 	WAIT_SYNCHRO_FOR_SLAVES();
 
+	rte_atomic32_set(&obj_count, 1); /* silent the check in the caller */
 	if (rte_eal_init(0, NULL) != -1)
 		return -1;
 
@@ -122,8 +124,8 @@ ring_create_lookup(__attribute__((unused)) void *arg)
 	/* create the same ring simultaneously on all threads */
 	for (i = 0; i < MAX_ITER_TIMES; i++) {
 		rp = rte_ring_create("fr_test_once", 4096, SOCKET_ID_ANY, 0);
-		if ((NULL == rp) && (rte_ring_lookup("fr_test_once") == NULL))
-			return -1;
+		if (rp != NULL)
+			rte_atomic32_inc(&obj_count);
 	}
 
 	/* create/lookup new ring several times */
@@ -172,8 +174,8 @@ mempool_create_lookup(__attribute__((unused)) void *arg)
 					NULL, NULL,
 					my_obj_init, NULL,
 					SOCKET_ID_ANY, 0);
-		if ((NULL == mp) && (rte_mempool_lookup("fr_test_once") == NULL))
-			return -1;
+		if (mp != NULL)
+			rte_atomic32_inc(&obj_count);
 	}
 
 	/* create/lookup new ring several times */
@@ -238,8 +240,8 @@ hash_create_free(__attribute__((unused)) void *arg)
 	hash_params.name = "fr_test_once";
 	for (i = 0; i < MAX_ITER_TIMES; i++) {
 		handle = rte_hash_create(&hash_params);
-		if ((NULL == handle) && (rte_hash_find_existing("fr_test_once") == NULL))
-			return -1;
+		if (handle != NULL)
+			rte_atomic32_inc(&obj_count);
 	}
 
 	/* create mutiple times simultaneously */
@@ -306,8 +308,8 @@ fbk_create_free(__attribute__((unused)) void *arg)
 	fbk_params.name = "fr_test_once";
 	for (i = 0; i < MAX_ITER_TIMES; i++) {
 		handle = rte_fbk_hash_create(&fbk_params);
-		if ((NULL == handle) && (rte_fbk_hash_find_existing("fr_test_once") == NULL))
-			return -1;
+		if (handle != NULL)
+			rte_atomic32_inc(&obj_count);
 	}
 
 	/* create mutiple fbk tables simultaneously */
@@ -372,8 +374,8 @@ lpm_create_free(__attribute__((unused)) void *arg)
 	/* create the same lpm simultaneously on all threads */
 	for (i = 0; i < MAX_ITER_TIMES; i++) {
 		lpm = rte_lpm_create("fr_test_once",  SOCKET_ID_ANY, &config);
-		if ((NULL == lpm) && (rte_lpm_find_existing("fr_test_once") == NULL))
-			return -1;
+		if (lpm != NULL)
+			rte_atomic32_inc(&obj_count);
 	}
 
 	/* create mutiple fbk tables simultaneously */
@@ -432,10 +434,12 @@ launch_test(struct test_case *pt_case)
 	unsigned lcore_id;
 	unsigned cores_save = rte_lcore_count();
 	unsigned cores = RTE_MIN(cores_save, MAX_LCORES);
+	unsigned count;
 
 	if (pt_case->func == NULL)
 		return -1;
 
+	rte_atomic32_set(&obj_count, 0);
 	rte_atomic32_set(&synchro, 0);
 
 	RTE_LCORE_FOREACH_SLAVE(lcore_id) {
@@ -462,6 +466,13 @@ launch_test(struct test_case *pt_case)
 			pt_case->clean(lcore_id);
 	}
 
+	count = rte_atomic32_read(&obj_count);
+	if (count != 1) {
+		printf("%s: common object allocated %d times (should be 1)\n",
+			pt_case->name, count);
+		ret = -1;
+	}
+
 	return ret;
 }
 
diff --git a/app/test/test_lpm6.c b/app/test/test_lpm6.c
index 1f88d7a..b464342 100644
--- a/app/test/test_lpm6.c
+++ b/app/test/test_lpm6.c
@@ -222,7 +222,7 @@ test1(void)
 
 	/* rte_lpm6_create: lpm name == LPM2 */
 	lpm3 = rte_lpm6_create("LPM1", SOCKET_ID_ANY, &config);
-	TEST_LPM_ASSERT(lpm3 == lpm1);
+	TEST_LPM_ASSERT(lpm3 == NULL);
 
 	rte_lpm6_free(lpm1);
 	rte_lpm6_free(lpm2);
diff --git a/lib/librte_hash/rte_cuckoo_hash.c b/lib/librte_hash/rte_cuckoo_hash.c
index 71b5b76..857e152 100644
--- a/lib/librte_hash/rte_cuckoo_hash.c
+++ b/lib/librte_hash/rte_cuckoo_hash.c
@@ -231,7 +231,7 @@ rte_hash_create(const struct rte_hash_parameters *params)
 	/* Guarantee there's no existing */
 	h = rte_hash_find_existing(params->name);
 	if (h != NULL)
-		return h;
+		return NULL;
 
 	te = rte_zmalloc("HASH_TAILQ_ENTRY", sizeof(*te), 0);
 	if (te == NULL) {
diff --git a/lib/librte_hash/rte_fbk_hash.c b/lib/librte_hash/rte_fbk_hash.c
index 8752a47..ea854f1 100644
--- a/lib/librte_hash/rte_fbk_hash.c
+++ b/lib/librte_hash/rte_fbk_hash.c
@@ -140,8 +140,10 @@ rte_fbk_hash_create(const struct rte_fbk_hash_params *params)
 		if (strncmp(params->name, ht->name, RTE_FBK_HASH_NAMESIZE) == 0)
 			break;
 	}
-	if (te != NULL)
+	if (te != NULL) {
+		ht = NULL;
 		goto exit;
+	}
 
 	te = rte_zmalloc("FBK_HASH_TAILQ_ENTRY", sizeof(*te), 0);
 	if (te == NULL) {
diff --git a/lib/librte_lpm/rte_lpm.c b/lib/librte_lpm/rte_lpm.c
index ccaaa2a..dd62f9b 100644
--- a/lib/librte_lpm/rte_lpm.c
+++ b/lib/librte_lpm/rte_lpm.c
@@ -209,8 +209,10 @@ rte_lpm_create_v20(const char *name, int socket_id, int max_rules,
 		if (strncmp(name, lpm->name, RTE_LPM_NAMESIZE) == 0)
 			break;
 	}
-	if (te != NULL)
+	if (te != NULL) {
+		lpm = NULL;
 		goto exit;
+	}
 
 	/* allocate tailq entry */
 	te = rte_zmalloc("LPM_TAILQ_ENTRY", sizeof(*te), 0);
@@ -280,8 +282,10 @@ rte_lpm_create_v1604(const char *name, int socket_id,
 		if (strncmp(name, lpm->name, RTE_LPM_NAMESIZE) == 0)
 			break;
 	}
-	if (te != NULL)
+	if (te != NULL) {
+		lpm = NULL;
 		goto exit;
+	}
 
 	/* allocate tailq entry */
 	te = rte_zmalloc("LPM_TAILQ_ENTRY", sizeof(*te), 0);
diff --git a/lib/librte_lpm/rte_lpm6.c b/lib/librte_lpm/rte_lpm6.c
index 6c2b293..4e9f2d0 100644
--- a/lib/librte_lpm/rte_lpm6.c
+++ b/lib/librte_lpm/rte_lpm6.c
@@ -182,8 +182,10 @@ rte_lpm6_create(const char *name, int socket_id,
 		if (strncmp(name, lpm->name, RTE_LPM6_NAMESIZE) == 0)
 			break;
 	}
-	if (te != NULL)
+	if (te != NULL) {
+		lpm = NULL;
 		goto exit;
+	}
 
 	/* allocate tailq entry */
 	te = rte_zmalloc("LPM6_TAILQ_ENTRY", sizeof(*te), 0);
-- 
2.1.4