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

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