linux-gen: stash: add mpmc ring based implementation

Use MPMC rings to implement overflow safe stash mode (application can
attempt to store more handles into a stash than it specified in the
creation parameters). This mode can be enabled by setting stash.strict_size
option to zero in the config file.

Signed-off-by: Matias Elo <matias.elo@nokia.com>
Reviewed-by: Tuomas Taipale <tuomas.taipale@nokia.com>

7 files changed, 140 insertions, 19 deletions

diff --git a/config/odp-linux-generic.conf b/config/odp-linux-generic.conf
index 3cb862ee4..df5328d1f 100644
--- a/config/odp-linux-generic.conf
+++ b/config/odp-linux-generic.conf
@@ -16,7 +16,7 @@
 
 # Mandatory fields
 odp_implementation = "linux-generic"
-config_file_version = "0.1.24"
+config_file_version = "0.1.25"
 
 # System options
 system: {
@@ -254,6 +254,12 @@ stash: {
 	# The value may be rounded up by the implementation. For optimal memory
 	# usage set value to a power of two - 1.
 	max_num_obj = 4095
+
+	# Strict size
+	#
+	# When set to 0, application can attempt to store more handles into a
+	# stash than it specified in the creation parameters.
+	strict_size = 1
 }
 
 timer: {
diff --git a/platform/linux-generic/m4/odp_libconfig.m4 b/platform/linux-generic/m4/odp_libconfig.m4
index d96a95055..0d268c935 100644
--- a/platform/linux-generic/m4/odp_libconfig.m4
+++ b/platform/linux-generic/m4/odp_libconfig.m4
@@ -3,7 +3,7 @@
 ##########################################################################
 m4_define([_odp_config_version_generation], [0])
 m4_define([_odp_config_version_major], [1])
-m4_define([_odp_config_version_minor], [24])
+m4_define([_odp_config_version_minor], [25])
 
 m4_define([_odp_config_version],
           [_odp_config_version_generation._odp_config_version_major._odp_config_version_minor])
diff --git a/platform/linux-generic/odp_stash.c b/platform/linux-generic/odp_stash.c
index a954cf866..5ff499843 100644
--- a/platform/linux-generic/odp_stash.c
+++ b/platform/linux-generic/odp_stash.c
@@ -18,6 +18,8 @@
 #include <odp_init_internal.h>
 #include <odp_libconfig_internal.h>
 #include <odp_macros_internal.h>
+#include <odp_ring_mpmc_u32_internal.h>
+#include <odp_ring_mpmc_u64_internal.h>
 #include <odp_ring_u32_internal.h>
 #include <odp_ring_u64_internal.h>
 
@@ -94,6 +96,16 @@ typedef struct ODP_ALIGNED_CACHE stash_t {
 			ring_u64_t hdr;
 			uint64_t   data[];
 		} ring_u64;
+
+		struct ODP_ALIGNED_CACHE {
+			ring_mpmc_u32_t hdr;
+			uint32_t   data[];
+		} ring_mpmc_u32;
+
+		struct ODP_ALIGNED_CACHE {
+			ring_mpmc_u64_t hdr;
+			uint64_t   data[];
+		} ring_mpmc_u64;
 	};
 
 } stash_t;
@@ -105,6 +117,7 @@ typedef struct stash_global_t {
 	uint32_t          max_num;
 	uint32_t          max_num_obj;
 	uint32_t          num_internal;
+	uint8_t           strict_size;
 	uint8_t           stash_state[CONFIG_MAX_STASHES];
 	stash_t           *stash[CONFIG_MAX_STASHES];
 	uint8_t           data[] ODP_ALIGNED_CACHE;
@@ -131,6 +144,7 @@ int _odp_stash_init_global(void)
 	uint64_t ring_max_size, stash_max_size, stash_data_size, offset;
 	const uint32_t internal_stashes = odp_global_ro.disable.dma ? 0 : CONFIG_INTERNAL_STASHES;
 	uint8_t *stash_data;
+	uint8_t strict_size;
 	int val = 0;
 
 	if (odp_global_ro.disable.stash && odp_global_ro.disable.dma) {
@@ -156,6 +170,14 @@ int _odp_stash_init_global(void)
 	_ODP_PRINT("  %s: %i\n", str, val);
 	max_num_obj = val;
 
+	str = "stash.strict_size";
+	if (!_odp_libconfig_lookup_int(str, &val)) {
+		_ODP_ERR("Config option '%s' not found.\n", str);
+		return -1;
+	}
+	_ODP_PRINT("  %s: %i\n", str, val);
+	strict_size = !!val;
+
 	_ODP_PRINT("\n");
 
 	/* Reserve resources for implementation internal stashes */
@@ -191,6 +213,7 @@ int _odp_stash_init_global(void)
 	stash_global->shm = shm;
 	stash_global->max_num = max_num;
 	stash_global->max_num_obj = max_num_obj;
+	stash_global->strict_size = strict_size;
 	stash_global->num_internal = internal_stashes;
 	odp_ticketlock_init(&stash_global->lock);
 
@@ -345,6 +368,80 @@ static inline int32_t strict_ring_u64_len(stash_t *stash)
 	return ring_u64_len(&stash->ring_u64.hdr);
 }
 
+static inline void mpmc_ring_u32_init(stash_t *stash)
+{
+	ring_mpmc_u32_init(&stash->ring_mpmc_u32.hdr);
+
+	for (uint32_t i = 0; i < stash->ring_size; i++)
+		stash->ring_mpmc_u32.data[i] = 0;
+}
+
+static inline void mpmc_ring_u64_init(stash_t *stash)
+{
+	ring_mpmc_u64_init(&stash->ring_mpmc_u64.hdr);
+
+	for (uint32_t i = 0; i < stash->ring_size; i++)
+		stash->ring_mpmc_u64.data[i] = 0;
+}
+
+static inline int32_t mpmc_ring_u32_enq_multi(stash_t *stash, const uint32_t val[], int32_t num)
+{
+	return ring_mpmc_u32_enq_multi(&stash->ring_mpmc_u32.hdr, stash->ring_mpmc_u32.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u64_enq_multi(stash_t *stash, const uint64_t val[], int32_t num)
+{
+	return ring_mpmc_u64_enq_multi(&stash->ring_mpmc_u64.hdr, stash->ring_mpmc_u64.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u32_enq_batch(stash_t *stash, const uint32_t val[], int32_t num)
+{
+	return ring_mpmc_u32_enq_batch(&stash->ring_mpmc_u32.hdr, stash->ring_mpmc_u32.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u64_enq_batch(stash_t *stash, const uint64_t val[], int32_t num)
+{
+	return ring_mpmc_u64_enq_batch(&stash->ring_mpmc_u64.hdr, stash->ring_mpmc_u64.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u32_deq_multi(stash_t *stash, uint32_t val[], int32_t num)
+{
+	return ring_mpmc_u32_deq_multi(&stash->ring_mpmc_u32.hdr, stash->ring_mpmc_u32.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u64_deq_multi(stash_t *stash, uint64_t val[], int32_t num)
+{
+	return ring_mpmc_u64_deq_multi(&stash->ring_mpmc_u64.hdr, stash->ring_mpmc_u64.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u32_deq_batch(stash_t *stash, uint32_t val[], int32_t num)
+{
+	return ring_mpmc_u32_deq_batch(&stash->ring_mpmc_u32.hdr, stash->ring_mpmc_u32.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u64_deq_batch(stash_t *stash, uint64_t val[], int32_t num)
+{
+	return ring_mpmc_u64_deq_batch(&stash->ring_mpmc_u64.hdr, stash->ring_mpmc_u64.data,
+				       stash->ring_mask, val, num);
+}
+
+static inline int32_t mpmc_ring_u32_len(stash_t *stash)
+{
+	return ring_mpmc_u32_len(&stash->ring_mpmc_u32.hdr);
+}
+
+static inline int32_t mpmc_ring_u64_len(stash_t *stash)
+{
+	return ring_mpmc_u64_len(&stash->ring_mpmc_u64.hdr);
+}
+
 odp_stash_t odp_stash_create(const char *name, const odp_stash_param_t *param)
 {
 	stash_t *stash;
@@ -394,20 +491,38 @@ odp_stash_t odp_stash_create(const char *name, const odp_stash_param_t *param)
 	memset(stash, 0, sizeof(stash_t));
 
 	/* Set ring function pointers */
-	if (ring_u64) {
-		stash->ring_fn.u64.init      = strict_ring_u64_init;
-		stash->ring_fn.u64.enq_multi = strict_ring_u64_enq_multi;
-		stash->ring_fn.u64.enq_batch = strict_ring_u64_enq_multi;
-		stash->ring_fn.u64.deq_multi = strict_ring_u64_deq_multi;
-		stash->ring_fn.u64.deq_batch = strict_ring_u64_deq_batch;
-		stash->ring_fn.u64.len       = strict_ring_u64_len;
+	if (stash_global->strict_size) {
+		if (ring_u64) {
+			stash->ring_fn.u64.init      = strict_ring_u64_init;
+			stash->ring_fn.u64.enq_multi = strict_ring_u64_enq_multi;
+			stash->ring_fn.u64.enq_batch = strict_ring_u64_enq_multi;
+			stash->ring_fn.u64.deq_multi = strict_ring_u64_deq_multi;
+			stash->ring_fn.u64.deq_batch = strict_ring_u64_deq_batch;
+			stash->ring_fn.u64.len       = strict_ring_u64_len;
+		} else {
+			stash->ring_fn.u32.init      = strict_ring_u32_init;
+			stash->ring_fn.u32.enq_multi = strict_ring_u32_enq_multi;
+			stash->ring_fn.u32.enq_batch = strict_ring_u32_enq_multi;
+			stash->ring_fn.u32.deq_multi = strict_ring_u32_deq_multi;
+			stash->ring_fn.u32.deq_batch = strict_ring_u32_deq_batch;
+			stash->ring_fn.u32.len       = strict_ring_u32_len;
+		}
 	} else {
-		stash->ring_fn.u32.init      = strict_ring_u32_init;
-		stash->ring_fn.u32.enq_multi = strict_ring_u32_enq_multi;
-		stash->ring_fn.u32.enq_batch = strict_ring_u32_enq_multi;
-		stash->ring_fn.u32.deq_multi = strict_ring_u32_deq_multi;
-		stash->ring_fn.u32.deq_batch = strict_ring_u32_deq_batch;
-		stash->ring_fn.u32.len       = strict_ring_u32_len;
+		if (ring_u64) {
+			stash->ring_fn.u64.init      = mpmc_ring_u64_init;
+			stash->ring_fn.u64.enq_multi = mpmc_ring_u64_enq_multi;
+			stash->ring_fn.u64.enq_batch = mpmc_ring_u64_enq_batch;
+			stash->ring_fn.u64.deq_multi = mpmc_ring_u64_deq_multi;
+			stash->ring_fn.u64.deq_batch = mpmc_ring_u64_deq_batch;
+			stash->ring_fn.u64.len       = mpmc_ring_u64_len;
+		} else {
+			stash->ring_fn.u32.init      = mpmc_ring_u32_init;
+			stash->ring_fn.u32.enq_multi = mpmc_ring_u32_enq_multi;
+			stash->ring_fn.u32.enq_batch = mpmc_ring_u32_enq_batch;
+			stash->ring_fn.u32.deq_multi = mpmc_ring_u32_deq_multi;
+			stash->ring_fn.u32.deq_batch = mpmc_ring_u32_deq_batch;
+			stash->ring_fn.u32.len       = mpmc_ring_u32_len;
+		}
 	}
 
 	if (name)
diff --git a/platform/linux-generic/test/inline-timer.conf b/platform/linux-generic/test/inline-timer.conf
index 56c1d40d1..44db4e337 100644
--- a/platform/linux-generic/test/inline-timer.conf
+++ b/platform/linux-generic/test/inline-timer.conf
@@ -1,6 +1,6 @@
 # Mandatory fields
 odp_implementation = "linux-generic"
-config_file_version = "0.1.24"
+config_file_version = "0.1.25"
 
 timer: {
 	# Enable inline timer implementation
diff --git a/platform/linux-generic/test/packet_align.conf b/platform/linux-generic/test/packet_align.conf
index 1aa7efc69..26491bd53 100644
--- a/platform/linux-generic/test/packet_align.conf
+++ b/platform/linux-generic/test/packet_align.conf
@@ -1,6 +1,6 @@
 # Mandatory fields
 odp_implementation = "linux-generic"
-config_file_version = "0.1.24"
+config_file_version = "0.1.25"
 
 pool: {
 	pkt: {
diff --git a/platform/linux-generic/test/process-mode.conf b/platform/linux-generic/test/process-mode.conf
index fb59a4f2a..2277aabdf 100644
--- a/platform/linux-generic/test/process-mode.conf
+++ b/platform/linux-generic/test/process-mode.conf
@@ -1,6 +1,6 @@
 # Mandatory fields
 odp_implementation = "linux-generic"
-config_file_version = "0.1.24"
+config_file_version = "0.1.25"
 
 # Shared memory options
 shm: {
diff --git a/platform/linux-generic/test/sched-basic.conf b/platform/linux-generic/test/sched-basic.conf
index f75141806..c9f7c79fd 100644
--- a/platform/linux-generic/test/sched-basic.conf
+++ b/platform/linux-generic/test/sched-basic.conf
@@ -1,6 +1,6 @@
 # Mandatory fields
 odp_implementation = "linux-generic"
-config_file_version = "0.1.24"
+config_file_version = "0.1.25"
 
 # Test scheduler with an odd spread value and without dynamic load balance
 sched_basic: {