[PATCH v6 13/21] pdcp: implement t-Reordering and packet buffering

[Anoob Joseph <anoobj@marvell.com>]

From: Volodymyr Fialko <vfialko@marvell.com>

Add in-order delivery of packets in PDCP. Delivery of packets in-order
relies on t-Reordering timer.

When 'out-of-order delivery' is disabled, PDCP will buffer all received
packets that are out of order. The t-Reordering timer determines the
time period these packets would be held in the buffer, waiting for any
missing packets to arrive.

Introduce packet buffering and state variables which indicate status of
the timer.

Signed-off-by: Anoob Joseph <anoobj@marvell.com>
Signed-off-by: Volodymyr Fialko <vfialko@marvell.com>
---
 lib/pdcp/meson.build    |   3 +-
 lib/pdcp/pdcp_entity.h  |  19 +++++++
 lib/pdcp/pdcp_process.c | 117 ++++++++++++++++++++++++++++++----------
 lib/pdcp/pdcp_reorder.c |  27 ++++++++++
 lib/pdcp/pdcp_reorder.h |  62 +++++++++++++++++++++
 lib/pdcp/rte_pdcp.c     |  53 ++++++++++++++++--
 lib/pdcp/rte_pdcp.h     |   6 ++-
 7 files changed, 252 insertions(+), 35 deletions(-)
 create mode 100644 lib/pdcp/pdcp_reorder.c
 create mode 100644 lib/pdcp/pdcp_reorder.h

diff --git a/lib/pdcp/meson.build b/lib/pdcp/meson.build
index 75d476bf6d..f4f9246bcb 100644
--- a/lib/pdcp/meson.build
+++ b/lib/pdcp/meson.build
@@ -12,9 +12,10 @@ sources = files(
         'pdcp_crypto.c',
         'pdcp_ctrl_pdu.c',
         'pdcp_process.c',
+        'pdcp_reorder.c',
         'rte_pdcp.c',
         )
 headers = files('rte_pdcp.h')
 indirect_headers += files('rte_pdcp_group.h')
 
-deps += ['mbuf', 'net', 'cryptodev', 'security']
+deps += ['mbuf', 'net', 'cryptodev', 'security', 'reorder']
diff --git a/lib/pdcp/pdcp_entity.h b/lib/pdcp/pdcp_entity.h
index 28691a504b..34341cdc11 100644
--- a/lib/pdcp/pdcp_entity.h
+++ b/lib/pdcp/pdcp_entity.h
@@ -11,6 +11,8 @@
 #include <rte_pdcp.h>
 #include <rte_security.h>
 
+#include "pdcp_reorder.h"
+
 struct entity_priv;
 
 #define PDCP_HFN_MIN 0
@@ -109,6 +111,17 @@ union cipher_iv_partial {
 	uint64_t u64[2];
 };
 
+enum timer_state {
+	TIMER_STOP,
+	TIMER_RUNNING,
+	TIMER_EXPIRED,
+};
+
+struct pdcp_t_reordering {
+	/** Represent timer state */
+	enum timer_state state;
+};
+
 struct pdcp_cnt_bitmap {
 	/** Number of entries that can be stored. */
 	uint32_t size;
@@ -145,6 +158,8 @@ struct entity_priv {
 		uint64_t is_null_auth : 1;
 		/** Is status report required.*/
 		uint64_t is_status_report_required : 1;
+		/** Is out-of-order delivery enabled */
+		uint64_t is_out_of_order_delivery : 1;
 	} flags;
 	/** Crypto op pool. */
 	struct rte_mempool *cop_pool;
@@ -161,6 +176,10 @@ struct entity_priv {
 struct entity_priv_dl_part {
 	/** PDCP would need to track the count values that are already received.*/
 	struct pdcp_cnt_bitmap bitmap;
+	/** t-Reordering handles */
+	struct pdcp_t_reordering t_reorder;
+	/** Reorder packet buffer */
+	struct pdcp_reorder reorder;
 };
 
 struct entity_priv_ul_part {
diff --git a/lib/pdcp/pdcp_process.c b/lib/pdcp/pdcp_process.c
index ed1413db6d..84a0f3a43f 100644
--- a/lib/pdcp/pdcp_process.c
+++ b/lib/pdcp/pdcp_process.c
@@ -837,25 +837,88 @@ pdcp_packet_strip(struct rte_mbuf *mb, const uint32_t hdr_trim_sz, const bool tr
 	}
 }
 
-static inline bool
+static inline int
 pdcp_post_process_update_entity_state(const struct rte_pdcp_entity *entity,
-				      const uint32_t count)
+				      const uint32_t count, struct rte_mbuf *mb,
+				      struct rte_mbuf *out_mb[],
+				      const bool trim_mac)
 {
 	struct entity_priv *en_priv = entity_priv_get(entity);
+	struct pdcp_t_reordering *t_reorder;
+	struct pdcp_reorder *reorder;
+	uint16_t processed = 0;
 
-	if (count < en_priv->state.rx_deliv)
-		return false;
+	struct entity_priv_dl_part *dl = entity_dl_part_get(entity);
+	const uint32_t hdr_trim_sz = en_priv->hdr_sz + en_priv->aad_sz;
 
-	/* t-Reordering timer is not supported - SDU will be delivered immediately.
-	 * Update RX_DELIV to the COUNT value of the first PDCP SDU which has not
-	 * been delivered to upper layers
-	 */
-	en_priv->state.rx_next = count + 1;
+	if (count < en_priv->state.rx_deliv)
+		return -EINVAL;
 
 	if (count >= en_priv->state.rx_next)
 		en_priv->state.rx_next = count + 1;
 
-	return true;
+	pdcp_packet_strip(mb, hdr_trim_sz, trim_mac);
+
+	if (en_priv->flags.is_out_of_order_delivery) {
+		out_mb[0] = mb;
+		en_priv->state.rx_deliv = count + 1;
+
+		return 1;
+	}
+
+	reorder = &dl->reorder;
+	t_reorder = &dl->t_reorder;
+
+	if (count == en_priv->state.rx_deliv) {
+		if (reorder->is_active) {
+			/*
+			 * This insert used only to increment reorder->min_seqn
+			 * To remove it - min_seqn_set() has to work with non-empty buffer
+			 */
+			pdcp_reorder_insert(reorder, mb, count);
+
+			/* Get buffered packets */
+			struct rte_mbuf **cached_mbufs = &out_mb[processed];
+			uint32_t nb_cached = pdcp_reorder_get_sequential(reorder,
+					cached_mbufs, entity->max_pkt_cache - processed);
+
+			processed += nb_cached;
+		} else {
+			out_mb[processed++] = mb;
+		}
+
+		/* Processed should never exceed the window size */
+		en_priv->state.rx_deliv = count + processed;
+
+	} else {
+		if (!reorder->is_active)
+			/* Initialize reordering buffer with RX_DELIV */
+			pdcp_reorder_start(reorder, en_priv->state.rx_deliv);
+		/* Buffer the packet */
+		pdcp_reorder_insert(reorder, mb, count);
+	}
+
+	/* Stop & reset current timer if rx_reord is received */
+	if (t_reorder->state == TIMER_RUNNING &&
+			en_priv->state.rx_deliv >= en_priv->state.rx_reord) {
+		t_reorder->state = TIMER_STOP;
+		/* Stop reorder buffer, only if it's empty */
+		if (en_priv->state.rx_deliv == en_priv->state.rx_next)
+			pdcp_reorder_stop(reorder);
+	}
+
+	/*
+	 * If t-Reordering is not running (includes the case when t-Reordering is stopped due to
+	 * actions above).
+	 */
+	if (t_reorder->state == TIMER_STOP && en_priv->state.rx_deliv < en_priv->state.rx_next) {
+		/* Update RX_REORD to RX_NEXT */
+		en_priv->state.rx_reord = en_priv->state.rx_next;
+		/* Start t-Reordering */
+		t_reorder->state = TIMER_RUNNING;
+	}
+
+	return processed;
 }
 
 static inline uint16_t
@@ -863,16 +926,12 @@ pdcp_post_process_uplane_dl_flags(const struct rte_pdcp_entity *entity, struct r
 				  struct rte_mbuf *out_mb[], uint16_t num, uint16_t *nb_err_ret,
 				  const bool is_integ_protected)
 {
-	struct entity_priv *en_priv = entity_priv_get(entity);
-	const uint32_t aad_sz = en_priv->aad_sz;
-	int i, nb_success = 0, nb_err = 0;
+	int i, nb_processed, nb_success = 0, nb_err = 0;
 	rte_pdcp_dynfield_t *mb_dynfield;
 	struct rte_mbuf *err_mb[num];
 	struct rte_mbuf *mb;
 	uint32_t count;
 
-	const uint32_t hdr_trim_sz = en_priv->hdr_sz + aad_sz;
-
 	for (i = 0; i < num; i++) {
 		mb = in_mb[i];
 		if (unlikely(mb->ol_flags & RTE_MBUF_F_RX_SEC_OFFLOAD_FAILED))
@@ -881,11 +940,12 @@ pdcp_post_process_uplane_dl_flags(const struct rte_pdcp_entity *entity, struct r
 		mb_dynfield = pdcp_dynfield(mb);
 		count = *mb_dynfield;
 
-		if (unlikely(!pdcp_post_process_update_entity_state(entity, count)))
+		nb_processed = pdcp_post_process_update_entity_state(
+				entity, count, mb, &out_mb[nb_success], is_integ_protected);
+		if (nb_processed < 0)
 			goto error;
 
-		pdcp_packet_strip(mb, hdr_trim_sz, is_integ_protected);
-		out_mb[nb_success++] = mb;
+		nb_success += nb_processed;
 		continue;
 
 error:
@@ -919,16 +979,12 @@ pdcp_post_process_cplane_sn_12_dl(const struct rte_pdcp_entity *entity,
 				  struct rte_mbuf *out_mb[],
 				  uint16_t num, uint16_t *nb_err_ret)
 {
-	struct entity_priv *en_priv = entity_priv_get(entity);
-	const uint32_t aad_sz = en_priv->aad_sz;
-	int i, nb_success = 0, nb_err = 0;
+	int i, nb_processed, nb_success = 0, nb_err = 0;
 	rte_pdcp_dynfield_t *mb_dynfield;
 	struct rte_mbuf *err_mb[num];
 	struct rte_mbuf *mb;
 	uint32_t count;
 
-	const uint32_t hdr_trim_sz = en_priv->hdr_sz + aad_sz;
-
 	for (i = 0; i < num; i++) {
 		mb = in_mb[i];
 		if (unlikely(mb->ol_flags & RTE_MBUF_F_RX_SEC_OFFLOAD_FAILED))
@@ -937,12 +993,12 @@ pdcp_post_process_cplane_sn_12_dl(const struct rte_pdcp_entity *entity,
 		mb_dynfield = pdcp_dynfield(mb);
 		count = *mb_dynfield;
 
-		if (unlikely(!pdcp_post_process_update_entity_state(entity, count)))
+		nb_processed = pdcp_post_process_update_entity_state(
+				entity, count, mb, &out_mb[nb_success], true);
+		if (nb_processed < 0)
 			goto error;
 
-		pdcp_packet_strip(mb, hdr_trim_sz, true);
-
-		out_mb[nb_success++] = mb;
+		nb_success += nb_processed;
 		continue;
 
 error:
@@ -1105,6 +1161,13 @@ pdcp_entity_priv_populate(struct entity_priv *en_priv, const struct rte_pdcp_ent
 		en_priv->flags.is_status_report_required = 1;
 	}
 
+	/**
+	 * flags.is_out_of_order_delivery
+	 *
+	 * Indicate whether the outoforder delivery is enabled for PDCP entity.
+	 */
+	en_priv->flags.is_out_of_order_delivery = conf->out_of_order_delivery;
+
 	/**
 	 * hdr_sz
 	 *
diff --git a/lib/pdcp/pdcp_reorder.c b/lib/pdcp/pdcp_reorder.c
new file mode 100644
index 0000000000..5399f0dc28
--- /dev/null
+++ b/lib/pdcp/pdcp_reorder.c
@@ -0,0 +1,27 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(C) 2023 Marvell.
+ */
+
+#include <rte_errno.h>
+#include <rte_reorder.h>
+
+#include "pdcp_reorder.h"
+
+int
+pdcp_reorder_create(struct pdcp_reorder *reorder, uint32_t window_size)
+{
+	reorder->buf = rte_reorder_create("reorder_buffer", SOCKET_ID_ANY, window_size);
+	if (reorder->buf == NULL)
+		return -rte_errno;
+
+	reorder->window_size = window_size;
+	reorder->is_active = false;
+
+	return 0;
+}
+
+void
+pdcp_reorder_destroy(const struct pdcp_reorder *reorder)
+{
+	rte_reorder_free(reorder->buf);
+}
diff --git a/lib/pdcp/pdcp_reorder.h b/lib/pdcp/pdcp_reorder.h
new file mode 100644
index 0000000000..6a2f61d6ae
--- /dev/null
+++ b/lib/pdcp/pdcp_reorder.h
@@ -0,0 +1,62 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(C) 2023 Marvell.
+ */
+
+#ifndef PDCP_REORDER_H
+#define PDCP_REORDER_H
+
+#include <rte_reorder.h>
+
+struct pdcp_reorder {
+	struct rte_reorder_buffer *buf;
+	uint32_t window_size;
+	bool is_active;
+};
+
+int pdcp_reorder_create(struct pdcp_reorder *reorder, uint32_t window_size);
+void pdcp_reorder_destroy(const struct pdcp_reorder *reorder);
+
+static inline uint32_t
+pdcp_reorder_get_sequential(struct pdcp_reorder *reorder, struct rte_mbuf **mbufs,
+		uint32_t max_mbufs)
+{
+	return rte_reorder_drain(reorder->buf, mbufs, max_mbufs);
+}
+
+static inline uint32_t
+pdcp_reorder_up_to_get(struct pdcp_reorder *reorder, struct rte_mbuf **mbufs,
+		       uint32_t max_mbufs, uint32_t seqn)
+{
+	return rte_reorder_drain_up_to_seqn(reorder->buf, mbufs, max_mbufs, seqn);
+}
+
+static inline void
+pdcp_reorder_start(struct pdcp_reorder *reorder, uint32_t min_seqn)
+{
+	int ret;
+
+	reorder->is_active = true;
+
+	ret = rte_reorder_min_seqn_set(reorder->buf, min_seqn);
+	RTE_VERIFY(ret == 0);
+}
+
+static inline void
+pdcp_reorder_stop(struct pdcp_reorder *reorder)
+{
+	reorder->is_active = false;
+}
+
+static inline void
+pdcp_reorder_insert(struct pdcp_reorder *reorder, struct rte_mbuf *mbuf,
+		    rte_reorder_seqn_t pkt_count)
+{
+	int ret;
+
+	*rte_reorder_seqn(mbuf) = pkt_count;
+
+	ret = rte_reorder_insert(reorder->buf, mbuf);
+	RTE_VERIFY(ret == 0);
+}
+
+#endif /* PDCP_REORDER_H */
diff --git a/lib/pdcp/rte_pdcp.c b/lib/pdcp/rte_pdcp.c
index 96ad397667..be37ff392c 100644
--- a/lib/pdcp/rte_pdcp.c
+++ b/lib/pdcp/rte_pdcp.c
@@ -49,6 +49,17 @@ pdcp_entity_size_get(const struct rte_pdcp_entity_conf *conf)
 	return RTE_ALIGN_CEIL(size, RTE_CACHE_LINE_SIZE);
 }
 
+static int
+pdcp_dl_establish(struct rte_pdcp_entity *entity, const struct rte_pdcp_entity_conf *conf)
+{
+	const uint32_t window_size = pdcp_window_size_get(conf->pdcp_xfrm.sn_size);
+	struct entity_priv_dl_part *dl = entity_dl_part_get(entity);
+
+	entity->max_pkt_cache = RTE_MAX(entity->max_pkt_cache, window_size);
+
+	return pdcp_reorder_create(&dl->reorder, window_size);
+}
+
 struct rte_pdcp_entity *
 rte_pdcp_entity_establish(const struct rte_pdcp_entity_conf *conf)
 {
@@ -118,6 +129,12 @@ rte_pdcp_entity_establish(const struct rte_pdcp_entity_conf *conf)
 	if (ret)
 		goto crypto_sess_destroy;
 
+	if (conf->pdcp_xfrm.pkt_dir == RTE_SECURITY_PDCP_DOWNLINK) {
+		ret = pdcp_dl_establish(entity, conf);
+		if (ret)
+			goto crypto_sess_destroy;
+	}
+
 	ret = pdcp_cnt_ring_create(entity, conf);
 	if (ret)
 		goto crypto_sess_destroy;
@@ -132,26 +149,50 @@ rte_pdcp_entity_establish(const struct rte_pdcp_entity_conf *conf)
 	return NULL;
 }
 
+static int
+pdcp_dl_release(struct rte_pdcp_entity *entity, struct rte_mbuf *out_mb[])
+{
+	struct entity_priv_dl_part *dl = entity_dl_part_get(entity);
+	struct entity_priv *en_priv = entity_priv_get(entity);
+	int nb_out;
+
+	nb_out = pdcp_reorder_up_to_get(&dl->reorder, out_mb, entity->max_pkt_cache,
+			en_priv->state.rx_next);
+
+	pdcp_reorder_destroy(&dl->reorder);
+
+	return nb_out;
+}
+
 int
 rte_pdcp_entity_release(struct rte_pdcp_entity *pdcp_entity, struct rte_mbuf *out_mb[])
 {
+	struct entity_priv *en_priv;
+	int nb_out = 0;
+
 	if (pdcp_entity == NULL)
 		return -EINVAL;
 
+	en_priv = entity_priv_get(pdcp_entity);
+
+	if (!en_priv->flags.is_ul_entity)
+		nb_out = pdcp_dl_release(pdcp_entity, out_mb);
+
 	/* Teardown crypto sessions */
 	pdcp_crypto_sess_destroy(pdcp_entity);
 
 	rte_free(pdcp_entity);
 
-	RTE_SET_USED(out_mb);
-	return 0;
+	return nb_out;
 }
 
 int
 rte_pdcp_entity_suspend(struct rte_pdcp_entity *pdcp_entity,
 			struct rte_mbuf *out_mb[])
 {
+	struct entity_priv_dl_part *dl;
 	struct entity_priv *en_priv;
+	int nb_out = 0;
 
 	if (pdcp_entity == NULL)
 		return -EINVAL;
@@ -161,13 +202,15 @@ rte_pdcp_entity_suspend(struct rte_pdcp_entity *pdcp_entity,
 	if (en_priv->flags.is_ul_entity) {
 		en_priv->state.tx_next = 0;
 	} else {
+		dl = entity_dl_part_get(pdcp_entity);
+		nb_out = pdcp_reorder_up_to_get(&dl->reorder, out_mb, pdcp_entity->max_pkt_cache,
+				en_priv->state.rx_next);
+		pdcp_reorder_stop(&dl->reorder);
 		en_priv->state.rx_next = 0;
 		en_priv->state.rx_deliv = 0;
 	}
 
-	RTE_SET_USED(out_mb);
-
-	return 0;
+	return nb_out;
 }
 
 struct rte_mbuf *
diff --git a/lib/pdcp/rte_pdcp.h b/lib/pdcp/rte_pdcp.h
index d7c2080358..9c4d06962a 100644
--- a/lib/pdcp/rte_pdcp.h
+++ b/lib/pdcp/rte_pdcp.h
@@ -110,6 +110,8 @@ struct rte_pdcp_entity_conf {
 	 * such as entity re-establishment.
 	 */
 	bool status_report_required;
+	/** Enable out of order delivery. */
+	bool out_of_order_delivery;
 };
 /* >8 End of structure rte_pdcp_entity_conf. */
 
@@ -269,8 +271,8 @@ rte_pdcp_pkt_pre_process(const struct rte_pdcp_entity *entity,
  * @param in_mb
  *   The address of an array of *num* pointers to *rte_mbuf* structures.
  * @param[out] out_mb
- *   The address of an array of *num* pointers to *rte_mbuf* structures
- *   to output packets after PDCP post-processing.
+ *   The address of an array that can hold up to *rte_pdcp_entity.max_pkt_cache*
+ *   pointers to *rte_mbuf* structures to output packets after PDCP post-processing.
  * @param num
  *   The maximum number of packets to process.
  * @param[out] nb_err
-- 
2.25.1