Merge branch 'dev/cleanup' of git://github.com/matiaselo/odp-dpdkv1.19.0.0_DPDK_17.11_tigermothv1.19.0.0_DPDK_17.11tigermoth_lts

Reviewed-by: Bill Fischofer <bill.fischofer@linaro.org>
Reviewed-by: Dmitry Eremin-Solenikov <dmitry.ereminsolenikov@linaro.org>
Signed-off-by: Maxim Uvarov <maxim.uvarov@linaro.org>

1 files changed, 106 insertions, 14 deletions

diff --git a/platform/linux-generic/odp_queue_basic.c b/platform/linux-generic/odp_queue_basic.c
index 1218987b1..89a0cd907 100644
--- a/platform/linux-generic/odp_queue_basic.c
+++ b/platform/linux-generic/odp_queue_basic.c
@@ -25,6 +25,7 @@
 #include <odp/api/hints.h>
 #include <odp/api/sync.h>
 #include <odp/api/traffic_mngr.h>
+#include <odp_libconfig_internal.h>
 
 #define NUM_INTERNAL_QUEUES 64
 
@@ -36,6 +37,9 @@
 #include <string.h>
 #include <inttypes.h>
 
+#define MIN_QUEUE_SIZE 8
+#define MAX_QUEUE_SIZE (1 * 1024 * 1024)
+
 static int queue_init(queue_entry_t *queue, const char *name,
 		      const odp_queue_param_t *param);
 
@@ -60,11 +64,11 @@ static int queue_capa(odp_queue_capability_t *capa, int sched)
 	/* Reserve some queues for internal use */
 	capa->max_queues        = ODP_CONFIG_QUEUES - NUM_INTERNAL_QUEUES;
 	capa->plain.max_num     = capa->max_queues;
-	capa->plain.max_size    = CONFIG_QUEUE_SIZE;
+	capa->plain.max_size    = queue_glb->config.max_queue_size;
 	capa->plain.lockfree.max_num  = queue_glb->queue_lf_num;
 	capa->plain.lockfree.max_size = queue_glb->queue_lf_size;
 	capa->sched.max_num     = capa->max_queues;
-	capa->sched.max_size    = CONFIG_QUEUE_SIZE;
+	capa->sched.max_size    = queue_glb->config.max_queue_size;
 
 	if (sched) {
 		capa->max_ordered_locks = sched_fn->max_ordered_locks();
@@ -75,6 +79,52 @@ static int queue_capa(odp_queue_capability_t *capa, int sched)
 	return 0;
 }
 
+static int read_config_file(queue_global_t *queue_glb)
+{
+	const char *str;
+	uint32_t val_u32;
+	int val = 0;
+
+	ODP_PRINT("Queue config:\n");
+
+	str = "queue_basic.max_queue_size";
+	if (!_odp_libconfig_lookup_int(str, &val)) {
+		ODP_ERR("Config option '%s' not found.\n", str);
+		return -1;
+	}
+
+	val_u32 = val;
+
+	if (val_u32 > MAX_QUEUE_SIZE || val_u32 < MIN_QUEUE_SIZE ||
+	    !CHECK_IS_POWER2(val_u32)) {
+		ODP_ERR("Bad value %s = %u\n", str, val_u32);
+		return -1;
+	}
+
+	queue_glb->config.max_queue_size = val_u32;
+	ODP_PRINT("  %s: %u\n", str, val_u32);
+
+	str = "queue_basic.default_queue_size";
+	if (!_odp_libconfig_lookup_int(str, &val)) {
+		ODP_ERR("Config option '%s' not found.\n", str);
+		return -1;
+	}
+
+	val_u32 = val;
+
+	if (val_u32 > queue_glb->config.max_queue_size ||
+	    val_u32 < MIN_QUEUE_SIZE ||
+	    !CHECK_IS_POWER2(val_u32)) {
+		ODP_ERR("Bad value %s = %u\n", str, val_u32);
+		return -1;
+	}
+
+	queue_glb->config.default_queue_size = val_u32;
+	ODP_PRINT("  %s: %u\n\n", str, val_u32);
+
+	return 0;
+}
+
 static int queue_init_global(void)
 {
 	uint32_t i;
@@ -82,10 +132,11 @@ static int queue_init_global(void)
 	uint32_t lf_size = 0;
 	queue_lf_func_t *lf_func;
 	odp_queue_capability_t capa;
+	uint64_t mem_size;
 
 	ODP_DBG("Starts...\n");
 
-	shm = odp_shm_reserve("odp_queues",
+	shm = odp_shm_reserve("_odp_queue_gbl",
 			      sizeof(queue_global_t),
 			      sizeof(queue_entry_t), 0);
 
@@ -104,6 +155,26 @@ static int queue_init_global(void)
 		queue->s.handle = queue_from_index(i);
 	}
 
+	if (read_config_file(queue_glb)) {
+		odp_shm_free(shm);
+		return -1;
+	}
+
+	queue_glb->queue_gbl_shm = shm;
+	mem_size = sizeof(uint32_t) * ODP_CONFIG_QUEUES *
+		   (uint64_t)queue_glb->config.max_queue_size;
+
+	shm = odp_shm_reserve("_odp_queue_rings", mem_size,
+			      ODP_CACHE_LINE_SIZE, 0);
+
+	if (shm == ODP_SHM_INVALID) {
+		odp_shm_free(queue_glb->queue_gbl_shm);
+		return -1;
+	}
+
+	queue_glb->queue_ring_shm = shm;
+	queue_glb->ring_data      = odp_shm_addr(shm);
+
 	lf_func = &queue_glb->queue_lf_func;
 	queue_glb->queue_lf_num  = queue_lf_init_global(&lf_size, lf_func);
 	queue_glb->queue_lf_size = lf_size;
@@ -133,7 +204,6 @@ static int queue_term_local(void)
 static int queue_term_global(void)
 {
 	int ret = 0;
-	int rc = 0;
 	queue_entry_t *queue;
 	int i;
 
@@ -142,20 +212,24 @@ static int queue_term_global(void)
 		LOCK(queue);
 		if (queue->s.status != QUEUE_STATUS_FREE) {
 			ODP_ERR("Not destroyed queue: %s\n", queue->s.name);
-			rc = -1;
+			ret = -1;
 		}
 		UNLOCK(queue);
 	}
 
 	queue_lf_term_global();
 
-	ret = odp_shm_free(odp_shm_lookup("odp_queues"));
-	if (ret < 0) {
-		ODP_ERR("shm free failed for odp_queues");
-		rc = -1;
+	if (odp_shm_free(queue_glb->queue_ring_shm)) {
+		ODP_ERR("shm free failed");
+		ret = -1;
 	}
 
-	return rc;
+	if (odp_shm_free(queue_glb->queue_gbl_shm)) {
+		ODP_ERR("shm free failed");
+		ret = -1;
+	}
+
+	return ret;
 }
 
 static int queue_capability(odp_queue_capability_t *capa)
@@ -207,7 +281,7 @@ static odp_queue_t queue_create(const char *name,
 	}
 
 	if (param->nonblocking == ODP_BLOCKING) {
-		if (param->size > CONFIG_QUEUE_SIZE)
+		if (param->size > queue_glb->config.max_queue_size)
 			return ODP_QUEUE_INVALID;
 	} else if (param->nonblocking == ODP_NONBLOCKING_LF) {
 		/* Only plain type lock-free queues supported */
@@ -586,6 +660,9 @@ static odp_event_t queue_deq(odp_queue_t handle)
 static int queue_init(queue_entry_t *queue, const char *name,
 		      const odp_queue_param_t *param)
 {
+	uint64_t offset;
+	uint32_t queue_size;
+
 	if (name == NULL) {
 		queue->s.name[0] = 0;
 	} else {
@@ -609,9 +686,24 @@ static int queue_init(queue_entry_t *queue, const char *name,
 	queue->s.pktin = PKTIN_INVALID;
 	queue->s.pktout = PKTOUT_INVALID;
 
-	ring_st_init(&queue->s.ring_st,
-		     queue_glb->ring_data[queue->s.index].data,
-		     CONFIG_QUEUE_SIZE);
+	/* Use default size for all small queues to quarantee performance
+	 * level. */
+	queue_size = queue_glb->config.default_queue_size;
+	if (param->size > queue_glb->config.default_queue_size)
+		queue_size = param->size;
+
+	/* Round up if not already a power of two */
+	queue_size = ROUNDUP_POWER2_U32(queue_size);
+
+	if (queue_size > queue_glb->config.max_queue_size) {
+		ODP_ERR("Too large queue size %u\n", queue_size);
+		return -1;
+	}
+
+	offset = queue->s.index * (uint64_t)queue_glb->config.max_queue_size;
+
+	ring_st_init(&queue->s.ring_st, &queue_glb->ring_data[offset],
+		     queue_size);
 
 	return 0;
 }