lpm: make tailq fully local

Since the data structures such as rings are shared in their entirety,
those TAILQ pointers are shared as well. Meaning that, after a
successful rte_ring creation, the tailq_next pointer of the last
ring in the TAILQ will be updated with a pointer to a ring which may
not be present in the address space of another process (i.e. a ring
that may be host-local or guest-local, and not shared over IVSHMEM).
Any successive ring create/lookup on the other side of IVSHMEM will
result in trying to dereference an invalid pointer.

This patchset fixes this problem by creating a default tailq entry
that may be used by any data structure that chooses to use TAILQs.
This default TAILQ entry will consist of a tailq_next/tailq_prev
pointers, and an opaque pointer to arbitrary data. All TAILQ
pointers from data structures themselves will be removed and
replaced by those generic TAILQ entries, thus fixing the problem
of potentially exposing local address space to shared structures.

Technically, only rte_ring structure require modification, because
IVSHMEM is only using memzones (which aren't in TAILQs) and rings,
but for consistency's sake other TAILQ-based data structures were
adapted as well.

Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
Acked-by: Konstantin Ananyev <konstantin.ananyev@intel.com>

lib/librte_lpm/rte_lpm.c

@@ -56,7 +56,7 @@
 
 #include "rte_lpm.h"
 
-TAILQ_HEAD(rte_lpm_list, rte_lpm);
+TAILQ_HEAD(rte_lpm_list, rte_tailq_entry);
 
 #define MAX_DEPTH_TBL24 24
 
@@ -118,24 +118,29 @@ depth_to_range(uint8_t depth)
 struct rte_lpm *
 rte_lpm_find_existing(const char *name)
 {
-	struct rte_lpm *l;
+	struct rte_lpm *l = NULL;
+	struct rte_tailq_entry *te;
 	struct rte_lpm_list *lpm_list;
 
 	/* check that we have an initialised tail queue */
-	if ((lpm_list = RTE_TAILQ_LOOKUP_BY_IDX(RTE_TAILQ_LPM, rte_lpm_list)) == NULL) {
+	if ((lpm_list = RTE_TAILQ_LOOKUP_BY_IDX(RTE_TAILQ_LPM,
+			rte_lpm_list)) == NULL) {
 		rte_errno = E_RTE_NO_TAILQ;
 		return NULL;
 	}
 
 	rte_rwlock_read_lock(RTE_EAL_TAILQ_RWLOCK);
-	TAILQ_FOREACH(l, lpm_list, next) {
+	TAILQ_FOREACH(te, lpm_list, next) {
+		l = (struct rte_lpm *) te->data;
 		if (strncmp(name, l->name, RTE_LPM_NAMESIZE) == 0)
 			break;
 	}
 	rte_rwlock_read_unlock(RTE_EAL_TAILQ_RWLOCK);
 
-	if (l == NULL)
+	if (te == NULL) {
 		rte_errno = ENOENT;
+		return NULL;
+	}
 
 	return l;
 }
@@ -149,12 +154,13 @@ rte_lpm_create(const char *name, int socket_id, int max_rules,
 {
 	char mem_name[RTE_LPM_NAMESIZE];
 	struct rte_lpm *lpm = NULL;
+	struct rte_tailq_entry *te;
 	uint32_t mem_size;
 	struct rte_lpm_list *lpm_list;
 
 	/* check that we have an initialised tail queue */
-	if ((lpm_list =
+	if ((lpm_list = RTE_TAILQ_LOOKUP_BY_IDX(RTE_TAILQ_LPM,
-	     RTE_TAILQ_LOOKUP_BY_IDX(RTE_TAILQ_LPM, rte_lpm_list)) == NULL) {
+			rte_lpm_list)) == NULL) {
 		rte_errno = E_RTE_NO_TAILQ;
 		return NULL;
 	}
@@ -176,18 +182,27 @@ rte_lpm_create(const char *name, int socket_id, int max_rules,
 	rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
 
 	/* guarantee there's no existing */
-	TAILQ_FOREACH(lpm, lpm_list, next) {
+	TAILQ_FOREACH(te, lpm_list, next) {
+		lpm = (struct rte_lpm *) te->data;
 		if (strncmp(name, lpm->name, RTE_LPM_NAMESIZE) == 0)
 			break;
 	}
-	if (lpm != NULL)
+	if (te != NULL)
 		goto exit;
 
+	/* allocate tailq entry */
+	te = rte_zmalloc("LPM_TAILQ_ENTRY", sizeof(*te), 0);
+	if (te == NULL) {
+		RTE_LOG(ERR, LPM, "Failed to allocate tailq entry\n");
+		goto exit;
+	}
+
 	/* Allocate memory to store the LPM data structures. */
 	lpm = (struct rte_lpm *)rte_zmalloc_socket(mem_name, mem_size,
 			CACHE_LINE_SIZE, socket_id);
 	if (lpm == NULL) {
 		RTE_LOG(ERR, LPM, "LPM memory allocation failed\n");
+		rte_free(te);
 		goto exit;
 	}
 
@@ -195,7 +210,9 @@ rte_lpm_create(const char *name, int socket_id, int max_rules,
 	lpm->max_rules = max_rules;
 	snprintf(lpm->name, sizeof(lpm->name), "%s", name);
 
-	TAILQ_INSERT_TAIL(lpm_list, lpm, next);
+	te->data = (void *) lpm;
+
+	TAILQ_INSERT_TAIL(lpm_list, te, next);
 
 exit:
 	rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
@@ -209,12 +226,38 @@ rte_lpm_create(const char *name, int socket_id, int max_rules,
 void
 rte_lpm_free(struct rte_lpm *lpm)
 {
+	struct rte_lpm_list *lpm_list;
+	struct rte_tailq_entry *te;
+
 	/* Check user arguments. */
 	if (lpm == NULL)
 		return;
 
-	RTE_EAL_TAILQ_REMOVE(RTE_TAILQ_LPM, rte_lpm_list, lpm);
+	/* check that we have an initialised tail queue */
+	if ((lpm_list =
+	     RTE_TAILQ_LOOKUP_BY_IDX(RTE_TAILQ_LPM, rte_lpm_list)) == NULL) {
+		rte_errno = E_RTE_NO_TAILQ;
+		return;
+	}
+
+	rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
+
+	/* find our tailq entry */
+	TAILQ_FOREACH(te, lpm_list, next) {
+		if (te->data == (void *) lpm)
+			break;
+	}
+	if (te == NULL) {
+		rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
+		return;
+	}
+
+	TAILQ_REMOVE(lpm_list, te, next);
+
+	rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
+
 	rte_free(lpm);
+	rte_free(te);
 }
 
 /*

lib/librte_lpm/rte_lpm.h

@@ -132,8 +132,6 @@ struct rte_lpm_rule_info {
 
 /** @internal LPM structure. */
 struct rte_lpm {
-	TAILQ_ENTRY(rte_lpm) next;      /**< Next in list. */
-
 	/* LPM metadata. */
 	char name[RTE_LPM_NAMESIZE];        /**< Name of the lpm. */
 	int mem_location; /**< @deprecated @see RTE_LPM_HEAP and RTE_LPM_MEMZONE. */

lib/librte_lpm/rte_lpm6.c

@@ -77,7 +77,7 @@ enum valid_flag {
 	VALID
 };
 
-TAILQ_HEAD(rte_lpm6_list, rte_lpm6);
+TAILQ_HEAD(rte_lpm6_list, rte_tailq_entry);
 
 /** Tbl entry structure. It is the same for both tbl24 and tbl8 */
 struct rte_lpm6_tbl_entry {
@@ -99,8 +99,6 @@ struct rte_lpm6_rule {
 
 /** LPM6 structure. */
 struct rte_lpm6 {
-	TAILQ_ENTRY(rte_lpm6) next;      /**< Next in list. */
-
 	/* LPM metadata. */
 	char name[RTE_LPM6_NAMESIZE];    /**< Name of the lpm. */
 	uint32_t max_rules;              /**< Max number of rules. */
@@ -149,6 +147,7 @@ rte_lpm6_create(const char *name, int socket_id,
 {
 	char mem_name[RTE_LPM6_NAMESIZE];
 	struct rte_lpm6 *lpm = NULL;
+	struct rte_tailq_entry *te;
 	uint64_t mem_size, rules_size;
 	struct rte_lpm6_list *lpm_list;
 
@@ -179,19 +178,28 @@ rte_lpm6_create(const char *name, int socket_id,
 	rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
 
 	/* Guarantee there's no existing */
-	TAILQ_FOREACH(lpm, lpm_list, next) {
+	TAILQ_FOREACH(te, lpm_list, next) {
+		lpm = (struct rte_lpm6 *) te->data;
 		if (strncmp(name, lpm->name, RTE_LPM6_NAMESIZE) == 0)
 			break;
 	}
-	if (lpm != NULL)
+	if (te != NULL)
 		goto exit;
 
+	/* allocate tailq entry */
+	te = rte_zmalloc("LPM6_TAILQ_ENTRY", sizeof(*te), 0);
+	if (te == NULL) {
+		RTE_LOG(ERR, LPM, "Failed to allocate tailq entry!\n");
+		goto exit;
+	}
+
 	/* Allocate memory to store the LPM data structures. */
 	lpm = (struct rte_lpm6 *)rte_zmalloc_socket(mem_name, (size_t)mem_size,
 			CACHE_LINE_SIZE, socket_id);
 
 	if (lpm == NULL) {
 		RTE_LOG(ERR, LPM, "LPM memory allocation failed\n");
+		rte_free(te);
 		goto exit;
 	}
 
@@ -201,6 +209,7 @@ rte_lpm6_create(const char *name, int socket_id,
 	if (lpm->rules_tbl == NULL) {
 		RTE_LOG(ERR, LPM, "LPM memory allocation failed\n");
 		rte_free(lpm);
+		rte_free(te);
 		goto exit;
 	}
 
@@ -209,7 +218,9 @@ rte_lpm6_create(const char *name, int socket_id,
 	lpm->number_tbl8s = config->number_tbl8s;
 	snprintf(lpm->name, sizeof(lpm->name), "%s", name);
 
-	TAILQ_INSERT_TAIL(lpm_list, lpm, next);
+	te->data = (void *) lpm;
+
+	TAILQ_INSERT_TAIL(lpm_list, te, next);
 
 exit:
 	rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
@@ -223,7 +234,8 @@ rte_lpm6_create(const char *name, int socket_id,
 struct rte_lpm6 *
 rte_lpm6_find_existing(const char *name)
 {
-	struct rte_lpm6 *l;
+	struct rte_lpm6 *l = NULL;
+	struct rte_tailq_entry *te;
 	struct rte_lpm6_list *lpm_list;
 
 	/* Check that we have an initialised tail queue */
@@ -234,14 +246,17 @@ rte_lpm6_find_existing(const char *name)
 	}
 
 	rte_rwlock_read_lock(RTE_EAL_TAILQ_RWLOCK);
-	TAILQ_FOREACH(l, lpm_list, next) {
+	TAILQ_FOREACH(te, lpm_list, next) {
+		l = (struct rte_lpm6 *) te->data;
 		if (strncmp(name, l->name, RTE_LPM6_NAMESIZE) == 0)
 			break;
 	}
 	rte_rwlock_read_unlock(RTE_EAL_TAILQ_RWLOCK);
 
-	if (l == NULL)
+	if (te == NULL) {
 		rte_errno = ENOENT;
+		return NULL;
+	}
 
 	return l;
 }
@@ -252,13 +267,38 @@ rte_lpm6_find_existing(const char *name)
 void
 rte_lpm6_free(struct rte_lpm6 *lpm)
 {
+	struct rte_lpm6_list *lpm_list;
+	struct rte_tailq_entry *te;
+
 	/* Check user arguments. */
 	if (lpm == NULL)
 		return;
 
-	RTE_EAL_TAILQ_REMOVE(RTE_TAILQ_LPM6, rte_lpm6_list, lpm);
+	/* check that we have an initialised tail queue */
-	rte_free(lpm->rules_tbl);
+	if ((lpm_list =
+	     RTE_TAILQ_LOOKUP_BY_IDX(RTE_TAILQ_LPM, rte_lpm6_list)) == NULL) {
+		rte_errno = E_RTE_NO_TAILQ;
+		return;
+	}
+
+	rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
+
+	/* find our tailq entry */
+	TAILQ_FOREACH(te, lpm_list, next) {
+		if (te->data == (void *) lpm)
+			break;
+	}
+	if (te == NULL) {
+		rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
+		return;
+	}
+
+	TAILQ_REMOVE(lpm_list, te, next);
+
+	rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
+
 	rte_free(lpm);
+	rte_free(te);
 }
 
 /*