linux-gen: ipsec: add configurable ordering mode

Add new config file parameters to control how original input queue
order is maintained during asynchronous IPsec processing.

When asynchronous IPsec operations are started within an ordered
scheduling context, packet order is maintained in IPsec completion
queues but the order in which the sequence number gets assigned or
the anti-replay check done is not guaranteed. This commit adds a config
option to force the stateful part of IPsec processing to be done in the
correct order, at the cost of reduced parallelism.

The ordering mechanism added in this commit uses the order_lock method
of the internal scheduling API to stop the calling thread until it
is its turn (i.e the preceding scheduling contexts associated with the
same input queue have been released). Future commits may add more
elaborate mechanisms.

The default behaviour does not change. By default no ordering is done.

Signed-off-by: Janne Peltonen <janne.peltonen@nokia.com>
Reviewed-by: Jere Leppänen <jere.leppanen@nokia.com>

8 files changed, 152 insertions, 28 deletions

diff --git a/config/odp-linux-generic.conf b/config/odp-linux-generic.conf
index 8469b72d7..3ac3e3e58 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.15"
+config_file_version = "0.1.16"
 
 # System options
 system: {
@@ -234,3 +234,52 @@ timer: {
 	# 2: Only control threads process non-private timer pools
 	inline_thread_type = 0
 }
+
+ipsec: {
+	# Packet ordering method for asynchronous IPsec processing
+	#
+	# Asynchronous IPsec processing maintains original packet order when
+	# started within ordered or atomic scheduling context. In addition
+	# to that, ODP API specifies that the order of IPsec processing
+	# (i.e. anti-replay window update and sequence number generation)
+	# is the same as the original packet order.
+	#
+	# The following settings control how the order is maintained in
+	# asynchronous IPsec operations. They have no effect on synchronous
+	# operations where the ODP application is responsible of the ordering.
+	#
+	# Values:
+	#
+	# 0: Ordering is not attempted.
+	#
+	#    This has the lowest overhead and the greatest parallelism but
+	#    is not fully compliant with the API specification.
+	#
+	#    Lack of ordering means that outbound IPsec packets, although
+	#    remaining in the correct order, may have their sequence numbers
+	#    assigned out of order. This can cause unexpected packet loss if
+	#    the anti-replay window of the receiving end is not large enough
+	#    to cover the possible misordering.
+	#
+	#    Similarly, since anti-replay check is not done in the reception
+	#    order, the anti-replay check sees additional packet misordering
+	#    on top of the true misordering of the received packets. This
+	#    means that a larger anti-replay window may be required to avoid
+	#    packet loss.
+	#
+	# 1: Ordering by waiting
+	#
+	#    Correct processing order is maintained by a simple mechanism
+	#    that makes a thread wait until its scheduling context has
+	#    reached the head of its input queue.
+	#
+	#    This limits parallelism when single input queue is used, even
+	#    when packets get distributed to multiple SAs.
+	ordering: {
+		  # Odering method for asynchronous inbound operations.
+		  async_inbound = 0
+
+		  # Odering method for asynchronous outbound operations.
+		  async_outbound = 0
+	}
+}
diff --git a/platform/linux-generic/m4/odp_libconfig.m4 b/platform/linux-generic/m4/odp_libconfig.m4
index 7a0b45497..ecfb28b7f 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], [15])
+m4_define([_odp_config_version_minor], [16])
 
 m4_define([_odp_config_version],
           [_odp_config_version_generation._odp_config_version_major._odp_config_version_minor])
diff --git a/platform/linux-generic/odp_ipsec.c b/platform/linux-generic/odp_ipsec.c
index e17e58a3f..bae1ecd12 100644
--- a/platform/linux-generic/odp_ipsec.c
+++ b/platform/linux-generic/odp_ipsec.c
@@ -19,6 +19,7 @@
 #include <odp_ipsec_internal.h>
 #include <odp/api/plat/queue_inlines.h>
 #include <odp_classification_internal.h>
+#include <odp_libconfig_internal.h>
 
 #include <protocols/eth.h>
 #include <protocols/ip.h>
@@ -28,9 +29,42 @@
 #include <errno.h>
 #include <string.h>
 
+typedef enum {
+	IPSEC_ORDERING_NONE = 0,
+	IPSEC_ORDERING_SIMPLE,
+} ordering_mode_t;
+
+typedef struct {
+	ordering_mode_t inbound_ordering_mode;
+	ordering_mode_t outbound_ordering_mode;
+	odp_ipsec_config_t ipsec_config;
+} ipsec_global_t;
+
+static ipsec_global_t *ipsec_global;
+
 static odp_ipsec_config_t *ipsec_config;
 
 /*
+ * Wait until the ordered scheduling context of this thread corresponds
+ * to the head of its input queue. Do nothing if ordering is not requested
+ * or if not holding an ordered context.
+ */
+static void wait_for_order(ordering_mode_t mode)
+{
+	if (mode == IPSEC_ORDERING_NONE)
+		return;
+	_odp_sched_fn->order_lock();
+	/*
+	 * We rely on the unlock being no-op, so let's not even bother
+	 * calling it. Unlock cannot really be anything but a no-op since
+	 * the scheduler cannot let other threads to continue until at
+	 * scheduling context release time.
+	 *
+	 * _odp_sched_fn->order_unlock();
+	 */
+}
+
+/*
  * Set cabability bits for algorithms that are defined for use with IPsec
  * and for which the IPsec crypto or auth capability function returns
  * at least one supported instance.
@@ -718,6 +752,7 @@ ipsec_sa_err_stats_update(ipsec_sa_t *sa, odp_ipsec_op_status_t *status)
 static ipsec_sa_t *ipsec_in_single(odp_packet_t pkt,
 				   odp_ipsec_sa_t sa,
 				   odp_packet_t *pkt_out,
+				   odp_bool_t enqueue_op,
 				   odp_ipsec_op_status_t *status)
 {
 	ipsec_state_t state;
@@ -804,10 +839,15 @@ static ipsec_sa_t *ipsec_in_single(odp_packet_t pkt,
 		goto exit;
 	}
 
-	if (_odp_ipsec_sa_replay_update(ipsec_sa,
-					state.in.seq_no,
-					status) < 0)
-		goto exit;
+	if (ipsec_sa->antireplay) {
+		if (enqueue_op)
+			wait_for_order(ipsec_global->inbound_ordering_mode);
+
+		if (_odp_ipsec_sa_replay_update(ipsec_sa,
+						state.in.seq_no,
+						status) < 0)
+			goto exit;
+	}
 
 	if (_odp_ipsec_sa_lifetime_update(ipsec_sa,
 					  state.stats_length,
@@ -1167,6 +1207,7 @@ static int ipsec_out_esp(odp_packet_t *pkt,
 			 odp_crypto_packet_op_param_t *param,
 			 odp_ipsec_op_status_t *status,
 			 uint32_t mtu,
+			 odp_bool_t enqueue_op,
 			 const odp_ipsec_out_opt_t *opt)
 {
 	_odp_esphdr_t esp;
@@ -1212,6 +1253,8 @@ static int ipsec_out_esp(odp_packet_t *pkt,
 		return -1;
 	}
 
+	if (enqueue_op)
+		wait_for_order(ipsec_global->outbound_ordering_mode);
 	seq_no = ipsec_seq_no(ipsec_sa);
 
 	if (ipsec_out_iv(state, ipsec_sa, seq_no) < 0) {
@@ -1339,7 +1382,8 @@ static int ipsec_out_ah(odp_packet_t *pkt,
 			ipsec_sa_t *ipsec_sa,
 			odp_crypto_packet_op_param_t *param,
 			odp_ipsec_op_status_t *status,
-			uint32_t mtu)
+			uint32_t mtu,
+			odp_bool_t enqueue_op)
 {
 	_odp_ahhdr_t ah;
 	unsigned hdr_len = _ODP_AHHDR_LEN + ipsec_sa->esp_iv_len +
@@ -1353,6 +1397,8 @@ static int ipsec_out_ah(odp_packet_t *pkt,
 		return -1;
 	}
 
+	if (enqueue_op)
+		wait_for_order(ipsec_global->outbound_ordering_mode);
 	seq_no = ipsec_seq_no(ipsec_sa);
 
 	memset(&ah, 0, sizeof(ah));
@@ -1497,6 +1543,7 @@ static ipsec_sa_t *ipsec_out_single(odp_packet_t pkt,
 				    odp_ipsec_sa_t sa,
 				    odp_packet_t *pkt_out,
 				    const odp_ipsec_out_opt_t *opt,
+				    odp_bool_t enqueue_op,
 				    odp_ipsec_op_status_t *status)
 {
 	ipsec_state_t state;
@@ -1602,9 +1649,10 @@ static ipsec_sa_t *ipsec_out_single(odp_packet_t pkt,
 
 	if (ODP_IPSEC_ESP == ipsec_sa->proto) {
 		rc = ipsec_out_esp(&pkt, &state, ipsec_sa, &param, status, mtu,
-				   opt);
+				   enqueue_op, opt);
 	} else if (ODP_IPSEC_AH == ipsec_sa->proto) {
-		rc = ipsec_out_ah(&pkt, &state, ipsec_sa, &param, status, mtu);
+		rc = ipsec_out_ah(&pkt, &state, ipsec_sa, &param, status, mtu,
+				  enqueue_op);
 	} else {
 		status->error.alg = 1;
 		goto exit;
@@ -1694,7 +1742,7 @@ int odp_ipsec_in(const odp_packet_t pkt_in[], int num_in,
 			ODP_ASSERT(ODP_IPSEC_SA_INVALID != sa);
 		}
 
-		ipsec_sa = ipsec_in_single(pkt, sa, &pkt, &status);
+		ipsec_sa = ipsec_in_single(pkt, sa, &pkt, false, &status);
 
 		packet_subtype_set(pkt, ODP_EVENT_PACKET_IPSEC);
 		result = ipsec_pkt_result(pkt);
@@ -1758,7 +1806,7 @@ int odp_ipsec_out(const odp_packet_t pkt_in[], int num_in,
 		else
 			opt = &param->opt[opt_idx];
 
-		ipsec_sa = ipsec_out_single(pkt, sa, &pkt, opt, &status);
+		ipsec_sa = ipsec_out_single(pkt, sa, &pkt, opt, false, &status);
 		ODP_ASSERT(NULL != ipsec_sa);
 
 		packet_subtype_set(pkt, ODP_EVENT_PACKET_IPSEC);
@@ -1806,7 +1854,7 @@ int odp_ipsec_in_enq(const odp_packet_t pkt_in[], int num_in,
 			ODP_ASSERT(ODP_IPSEC_SA_INVALID != sa);
 		}
 
-		ipsec_sa = ipsec_in_single(pkt, sa, &pkt, &status);
+		ipsec_sa = ipsec_in_single(pkt, sa, &pkt, true, &status);
 
 		packet_subtype_set(pkt, ODP_EVENT_PACKET_IPSEC);
 		result = ipsec_pkt_result(pkt);
@@ -1869,7 +1917,7 @@ int odp_ipsec_out_enq(const odp_packet_t pkt_in[], int num_in,
 		else
 			opt = &param->opt[opt_idx];
 
-		ipsec_sa = ipsec_out_single(pkt, sa, &pkt, opt, &status);
+		ipsec_sa = ipsec_out_single(pkt, sa, &pkt, opt, true, &status);
 		ODP_ASSERT(NULL != ipsec_sa);
 
 		packet_subtype_set(pkt, ODP_EVENT_PACKET_IPSEC);
@@ -1906,7 +1954,8 @@ int _odp_ipsec_try_inline(odp_packet_t *pkt)
 
 	memset(&status, 0, sizeof(status));
 
-	ipsec_sa = ipsec_in_single(*pkt, ODP_IPSEC_SA_INVALID, pkt, &status);
+	ipsec_sa = ipsec_in_single(*pkt, ODP_IPSEC_SA_INVALID, pkt, false,
+				   &status);
 	/*
 	 * Route packet back in case of lookup failure or early error before
 	 * lookup
@@ -1993,7 +2042,7 @@ int odp_ipsec_out_inline(const odp_packet_t pkt_in[], int num_in,
 		else
 			opt = &param->opt[opt_idx];
 
-		ipsec_sa = ipsec_out_single(pkt, sa, &pkt, opt, &status);
+		ipsec_sa = ipsec_out_single(pkt, sa, &pkt, opt, true, &status);
 		ODP_ASSERT(NULL != ipsec_sa);
 
 		offset = odp_packet_l3_offset(pkt);
@@ -2138,6 +2187,27 @@ int odp_ipsec_stats(odp_ipsec_sa_t sa, odp_ipsec_stats_t *stats)
 	return 0;
 }
 
+static int read_config_file(ipsec_global_t *global)
+{
+	const char *str_i = "ipsec.ordering.async_inbound";
+	const char *str_o = "ipsec.ordering.async_outbound";
+	int val;
+
+	if (!_odp_libconfig_lookup_int(str_i, &val)) {
+		ODP_ERR("Config option '%s' not found.\n", str_i);
+		return -1;
+	}
+	global->inbound_ordering_mode = val;
+
+	if (!_odp_libconfig_lookup_int(str_o, &val)) {
+		ODP_ERR("Config option '%s' not found.\n", str_o);
+		return -1;
+	}
+	global->outbound_ordering_mode = val;
+
+	return 0;
+}
+
 int _odp_ipsec_init_global(void)
 {
 	odp_shm_t shm;
@@ -2145,17 +2215,25 @@ int _odp_ipsec_init_global(void)
 	if (odp_global_ro.disable.ipsec)
 		return 0;
 
-	shm = odp_shm_reserve("_odp_ipsec", sizeof(odp_ipsec_config_t),
+	shm = odp_shm_reserve("_odp_ipsec", sizeof(*ipsec_global),
 			      ODP_CACHE_LINE_SIZE, 0);
-
-	ipsec_config = odp_shm_addr(shm);
-
-	if (ipsec_config == NULL) {
+	if (shm == ODP_SHM_INVALID) {
 		ODP_ERR("Shm reserve failed for odp_ipsec\n");
 		return -1;
 	}
+	ipsec_global = odp_shm_addr(shm);
+	if (ipsec_global == NULL) {
+		ODP_ERR("ipsec: odp_shm_addr() failed\n");
+		odp_shm_free(shm);
+		return -1;
+	}
+	memset(ipsec_global, 0, sizeof(*ipsec_global));
+	ipsec_config = &ipsec_global->ipsec_config;
 
-	odp_ipsec_config_init(ipsec_config);
+	if (read_config_file(ipsec_global)) {
+		odp_shm_free(shm);
+		return -1;
+	}
 
 	memset(&default_out_opt, 0, sizeof(default_out_opt));
 
diff --git a/platform/linux-generic/odp_ipsec_sad.c b/platform/linux-generic/odp_ipsec_sad.c
index 0ebecee7c..7a79e83d2 100644
--- a/platform/linux-generic/odp_ipsec_sad.c
+++ b/platform/linux-generic/odp_ipsec_sad.c
@@ -932,9 +932,6 @@ int _odp_ipsec_sa_replay_update(ipsec_sa_t *ipsec_sa, uint32_t seq,
 	int cas = 0;
 	uint64_t state, new_state;
 
-	if (!ipsec_sa->antireplay)
-		return 0;
-
 	state = odp_atomic_load_u64(&ipsec_sa->hot.in.antireplay);
 
 	while (0 == cas) {
diff --git a/platform/linux-generic/test/inline-timer.conf b/platform/linux-generic/test/inline-timer.conf
index e4d4af307..93195f5a8 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.15"
+config_file_version = "0.1.16"
 
 timer: {
 	# Enable inline timer implementation
diff --git a/platform/linux-generic/test/packet_align.conf b/platform/linux-generic/test/packet_align.conf
index 9b37752ba..58a73f2df 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.15"
+config_file_version = "0.1.16"
 
 pool: {
 	pkt: {
diff --git a/platform/linux-generic/test/process-mode.conf b/platform/linux-generic/test/process-mode.conf
index 5354cae2f..a6e6080d2 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.15"
+config_file_version = "0.1.16"
 
 # Shared memory options
 shm: {
diff --git a/platform/linux-generic/test/sched-basic.conf b/platform/linux-generic/test/sched-basic.conf
index 57a8a772c..79537b454 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.15"
+config_file_version = "0.1.16"
 
 sched_basic: {
 	# Test scheduler with an odd spread value