[dpdk-dev] [PATCH v1 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[dpdk-dev] [PATCH v1 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[Piotr Azarewicz]

This patch set enhancement ip_pipeline application:
- librte_port: add support for multi-producer/multi-consumer ring ports
- librte_port: bug fixes for ring ports with IPv4/IPv6 reassembly support
- ip_pipeline application: integrate MP/MC and fragmentation/reassembly support
 to SWQs

Piotr Azarewicz (3):
  port: add mp/mc ring ports
  port: fix ras ring ports
  examples/ip_pipeline: add mp/mc and frag/ras swq

 examples/ip_pipeline/app.h          |   14 ++
 examples/ip_pipeline/config_check.c |   45 +++-
 examples/ip_pipeline/config_parse.c |  195 ++++++++++++++++-
 examples/ip_pipeline/init.c         |  165 ++++++++++++---
 examples/ip_pipeline/main.c         |    4 +-
 examples/ip_pipeline/pipeline_be.h  |   18 ++
 lib/librte_port/rte_port_ras.c      |    8 +-
 lib/librte_port/rte_port_ring.c     |  399 ++++++++++++++++++++++++++++++++++-
 lib/librte_port/rte_port_ring.h     |   34 ++-
 9 files changed, 832 insertions(+), 50 deletions(-)

-- 
1.7.9.5

[dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Piotr Azarewicz]

ring_multi_reader input port (on top of multi consumer rte_ring)
ring_multi_writer output port (on top of multi producer rte_ring)

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_ring.c |  399 ++++++++++++++++++++++++++++++++++++++-
 lib/librte_port/rte_port_ring.h |   34 +++-
 2 files changed, 424 insertions(+), 9 deletions(-)

diff --git a/lib/librte_port/rte_port_ring.c b/lib/librte_port/rte_port_ring.c
index 9461c05..6b06466 100644
--- a/lib/librte_port/rte_port_ring.c
+++ b/lib/librte_port/rte_port_ring.c
@@ -70,8 +70,10 @@ rte_port_ring_reader_create(void *params, int socket_id)
 	struct rte_port_ring_reader *port;
 
 	/* Check input parameters */
-	if (conf == NULL) {
-		RTE_LOG(ERR, PORT, "%s: params is NULL\n", __func__);
+	if ((conf == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->cons.sc_dequeue != 1)) {
+		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
 	}
 
@@ -166,7 +168,8 @@ rte_port_ring_writer_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue != 1) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -343,7 +346,8 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue != 1) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -448,6 +452,7 @@ rte_port_ring_writer_nodrop_tx_bulk(void *port,
 		 */
 		for (; n_pkts_ok < n_pkts; n_pkts_ok++) {
 			struct rte_mbuf *pkt = pkts[n_pkts_ok];
+
 			p->tx_buf[p->tx_buf_count++] = pkt;
 		}
 		send_burst_nodrop(p);
@@ -513,6 +518,367 @@ rte_port_ring_writer_nodrop_stats_read(void *port,
 }
 
 /*
+ * Port RING Multi Reader
+ */
+static void *
+rte_port_ring_multi_reader_create(void *params, int socket_id)
+{
+	struct rte_port_ring_multi_reader_params *conf =
+			(struct rte_port_ring_multi_reader_params *) params;
+	struct rte_port_ring_reader *port;
+
+	/* Check input parameters */
+	if ((conf == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->cons.sc_dequeue != 0)) {
+		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
+		return NULL;
+	}
+
+	/* Memory allocation */
+	port = rte_zmalloc_socket("PORT", sizeof(*port),
+			RTE_CACHE_LINE_SIZE, socket_id);
+	if (port == NULL) {
+		RTE_LOG(ERR, PORT, "%s: Failed to allocate port\n", __func__);
+		return NULL;
+	}
+
+	/* Initialization */
+	port->ring = conf->ring;
+
+	return port;
+}
+
+static int
+rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
+{
+	struct rte_port_ring_reader *p = (struct rte_port_ring_reader *) port;
+	uint32_t nb_rx;
+
+	nb_rx = rte_ring_mc_dequeue_burst(p->ring, (void **) pkts, n_pkts);
+	RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx);
+
+	return nb_rx;
+}
+
+static int
+rte_port_ring_multi_reader_free(void *port)
+{
+	if (port == NULL) {
+		RTE_LOG(ERR, PORT, "%s: port is NULL\n", __func__);
+		return -EINVAL;
+	}
+
+	rte_free(port);
+
+	return 0;
+}
+
+/*
+ * Port RING Multi Writer
+ */
+static void *
+rte_port_ring_multi_writer_create(void *params, int socket_id)
+{
+	struct rte_port_ring_multi_writer_params *conf =
+			(struct rte_port_ring_multi_writer_params *) params;
+	struct rte_port_ring_writer *port;
+
+	/* Check input parameters */
+	if ((conf == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue != 0) ||
+		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
+		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
+		return NULL;
+	}
+
+	/* Memory allocation */
+	port = rte_zmalloc_socket("PORT", sizeof(*port),
+			RTE_CACHE_LINE_SIZE, socket_id);
+	if (port == NULL) {
+		RTE_LOG(ERR, PORT, "%s: Failed to allocate port\n", __func__);
+		return NULL;
+	}
+
+	/* Initialization */
+	port->ring = conf->ring;
+	port->tx_burst_sz = conf->tx_burst_sz;
+	port->tx_buf_count = 0;
+	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+
+	return port;
+}
+
+static inline void
+send_burst_mp(struct rte_port_ring_writer *p)
+{
+	uint32_t nb_tx;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+			p->tx_buf_count);
+
+	RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
+static int
+rte_port_ring_multi_writer_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static int
+rte_port_ring_multi_writer_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	struct rte_port_ring_writer *p =
+		(struct rte_port_ring_writer *) port;
+
+	uint32_t bsz_mask = p->bsz_mask;
+	uint32_t tx_buf_count = p->tx_buf_count;
+	uint64_t expr = (pkts_mask & (pkts_mask + 1)) |
+			((pkts_mask & bsz_mask) ^ bsz_mask);
+
+	if (expr == 0) {
+		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
+		uint32_t n_pkts_ok;
+
+		if (tx_buf_count)
+			send_burst_mp(p);
+
+		RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, n_pkts);
+		n_pkts_ok = rte_ring_mp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+
+		RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, n_pkts - n_pkts_ok);
+		for ( ; n_pkts_ok < n_pkts; n_pkts_ok++) {
+			struct rte_mbuf *pkt = pkts[n_pkts_ok];
+
+			rte_pktmbuf_free(pkt);
+		}
+	} else {
+		for ( ; pkts_mask; ) {
+			uint32_t pkt_index = __builtin_ctzll(pkts_mask);
+			uint64_t pkt_mask = 1LLU << pkt_index;
+			struct rte_mbuf *pkt = pkts[pkt_index];
+
+			p->tx_buf[tx_buf_count++] = pkt;
+			RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
+			pkts_mask &= ~pkt_mask;
+		}
+
+		p->tx_buf_count = tx_buf_count;
+		if (tx_buf_count >= p->tx_burst_sz)
+			send_burst_mp(p);
+	}
+
+	return 0;
+}
+
+static int
+rte_port_ring_multi_writer_flush(void *port)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static int
+rte_port_ring_multi_writer_free(void *port)
+{
+	if (port == NULL) {
+		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
+		return -EINVAL;
+	}
+
+	rte_port_ring_multi_writer_flush(port);
+	rte_free(port);
+
+	return 0;
+}
+
+/*
+ * Port RING Multi Writer Nodrop
+ */
+static void *
+rte_port_ring_multi_writer_nodrop_create(void *params, int socket_id)
+{
+	struct rte_port_ring_multi_writer_nodrop_params *conf =
+			(struct rte_port_ring_multi_writer_nodrop_params *) params;
+	struct rte_port_ring_writer_nodrop *port;
+
+	/* Check input parameters */
+	if ((conf == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue != 0) ||
+		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
+		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
+		return NULL;
+	}
+
+	/* Memory allocation */
+	port = rte_zmalloc_socket("PORT", sizeof(*port),
+			RTE_CACHE_LINE_SIZE, socket_id);
+	if (port == NULL) {
+		RTE_LOG(ERR, PORT, "%s: Failed to allocate port\n", __func__);
+		return NULL;
+	}
+
+	/* Initialization */
+	port->ring = conf->ring;
+	port->tx_burst_sz = conf->tx_burst_sz;
+	port->tx_buf_count = 0;
+	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+
+	/*
+	 * When n_retries is 0 it means that we should wait for every packet to
+	 * send no matter how many retries should it take. To limit number of
+	 * branches in fast path, we use UINT64_MAX instead of branching.
+	 */
+	port->n_retries = (conf->n_retries == 0) ? UINT64_MAX : conf->n_retries;
+
+	return port;
+}
+
+static inline void
+send_burst_mp_nodrop(struct rte_port_ring_writer_nodrop *p)
+{
+	uint32_t nb_tx = 0, i;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+				p->tx_buf_count);
+
+	/* We sent all the packets in a first try */
+	if (nb_tx >= p->tx_buf_count)
+		return;
+
+	for (i = 0; i < p->n_retries; i++) {
+		nb_tx += rte_ring_mp_enqueue_burst(p->ring,
+				(void **) (p->tx_buf + nb_tx), p->tx_buf_count - nb_tx);
+
+		/* We sent all the packets in more than one try */
+		if (nb_tx >= p->tx_buf_count)
+			return;
+	}
+
+	/* We didn't send the packets in maximum allowed attempts */
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
+static int
+rte_port_ring_multi_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static int
+rte_port_ring_multi_writer_nodrop_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	struct rte_port_ring_writer_nodrop *p =
+		(struct rte_port_ring_writer_nodrop *) port;
+
+	uint32_t bsz_mask = p->bsz_mask;
+	uint32_t tx_buf_count = p->tx_buf_count;
+	uint64_t expr = (pkts_mask & (pkts_mask + 1)) |
+			((pkts_mask & bsz_mask) ^ bsz_mask);
+
+	if (expr == 0) {
+		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
+		uint32_t n_pkts_ok;
+
+		if (tx_buf_count)
+			send_burst_mp_nodrop(p);
+
+		RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, n_pkts);
+		n_pkts_ok = rte_ring_mp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+
+		if (n_pkts_ok >= n_pkts)
+			return 0;
+
+		/*
+		 * If we didnt manage to send all packets in single burst, move
+		 * remaining packets to the buffer and call send burst.
+		 */
+		for (; n_pkts_ok < n_pkts; n_pkts_ok++) {
+			struct rte_mbuf *pkt = pkts[n_pkts_ok];
+
+			p->tx_buf[p->tx_buf_count++] = pkt;
+		}
+		send_burst_mp_nodrop(p);
+	} else {
+		for ( ; pkts_mask; ) {
+			uint32_t pkt_index = __builtin_ctzll(pkts_mask);
+			uint64_t pkt_mask = 1LLU << pkt_index;
+			struct rte_mbuf *pkt = pkts[pkt_index];
+
+			p->tx_buf[tx_buf_count++] = pkt;
+			RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
+			pkts_mask &= ~pkt_mask;
+		}
+
+		p->tx_buf_count = tx_buf_count;
+		if (tx_buf_count >= p->tx_burst_sz)
+			send_burst_mp_nodrop(p);
+	}
+
+	return 0;
+}
+
+static int
+rte_port_ring_multi_writer_nodrop_flush(void *port)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static int
+rte_port_ring_multi_writer_nodrop_free(void *port)
+{
+	if (port == NULL) {
+		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
+		return -EINVAL;
+	}
+
+	rte_port_ring_multi_writer_nodrop_flush(port);
+	rte_free(port);
+
+	return 0;
+}
+
+/*
  * Summary of port operations
  */
 struct rte_port_in_ops rte_port_ring_reader_ops = {
@@ -539,3 +905,28 @@ struct rte_port_out_ops rte_port_ring_writer_nodrop_ops = {
 	.f_flush = rte_port_ring_writer_nodrop_flush,
 	.f_stats = rte_port_ring_writer_nodrop_stats_read,
 };
+
+struct rte_port_in_ops rte_port_ring_multi_reader_ops = {
+	.f_create = rte_port_ring_multi_reader_create,
+	.f_free = rte_port_ring_multi_reader_free,
+	.f_rx = rte_port_ring_multi_reader_rx,
+	.f_stats = rte_port_ring_reader_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_ops = {
+	.f_create = rte_port_ring_multi_writer_create,
+	.f_free = rte_port_ring_multi_writer_free,
+	.f_tx = rte_port_ring_multi_writer_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_flush,
+	.f_stats = rte_port_ring_writer_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops = {
+	.f_create = rte_port_ring_multi_writer_nodrop_create,
+	.f_free = rte_port_ring_multi_writer_nodrop_free,
+	.f_tx = rte_port_ring_multi_writer_nodrop_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_nodrop_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_nodrop_flush,
+	.f_stats = rte_port_ring_writer_nodrop_stats_read,
+};
diff --git a/lib/librte_port/rte_port_ring.h b/lib/librte_port/rte_port_ring.h
index 89a219b..6999340 100644
--- a/lib/librte_port/rte_port_ring.h
+++ b/lib/librte_port/rte_port_ring.h
@@ -42,8 +42,14 @@ extern "C" {
  * @file
  * RTE Port Ring
  *
- * ring_reader: input port built on top of pre-initialized single consumer ring
- * ring_writer: output port built on top of pre-initialized single producer ring
+ * ring_reader:
+ *      input port built on top of pre-initialized single consumer ring
+ * ring_writer:
+ *      output port built on top of pre-initialized single producer ring
+ * ring_multi_reader:
+ *      input port built on top of pre-initialized multi consumers ring
+ * ring_multi_writer:
+ *      output port built on top of pre-initialized multi producers ring
  *
  ***/
 
@@ -55,7 +61,7 @@ extern "C" {
 
 /** ring_reader port parameters */
 struct rte_port_ring_reader_params {
-	/** Underlying single consumer ring that has to be pre-initialized */
+	/** Underlying consumer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 };
 
@@ -64,7 +70,7 @@ extern struct rte_port_in_ops rte_port_ring_reader_ops;
 
 /** ring_writer port parameters */
 struct rte_port_ring_writer_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -77,7 +83,7 @@ extern struct rte_port_out_ops rte_port_ring_writer_ops;
 
 /** ring_writer_nodrop port parameters */
 struct rte_port_ring_writer_nodrop_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -91,6 +97,24 @@ struct rte_port_ring_writer_nodrop_params {
 /** ring_writer_nodrop port operations */
 extern struct rte_port_out_ops rte_port_ring_writer_nodrop_ops;
 
+/** ring_multi_reader port parameters */
+#define rte_port_ring_multi_reader_params rte_port_ring_reader_params
+
+/** ring_multi_reader port operations */
+extern struct rte_port_in_ops rte_port_ring_multi_reader_ops;
+
+/** ring_multi_writer port parameters */
+#define rte_port_ring_multi_writer_params rte_port_ring_writer_params
+
+/** ring_multi_writer port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_ops;
+
+/** ring_multi_writer_nodrop port parameters */
+#define rte_port_ring_multi_writer_nodrop_params rte_port_ring_writer_nodrop_params
+
+/** ring_multi_writer_nodrop port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops;
+
 #ifdef __cplusplus
 }
 #endif
-- 
1.7.9.5

[dpdk-dev] [PATCH v1 2/3] port: fix ras ring ports

[Piotr Azarewicz]

Bug fixes for ring ports with IPv4/IPv6 reassembly support.
Previous implementation can't work properly due to incorrect choosing
process function.
Also, assuming that, when processing ip packet, ip header is know we can
set l3_len parameter here.

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_ras.c |    8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/lib/librte_port/rte_port_ras.c b/lib/librte_port/rte_port_ras.c
index 6bd0f8c..e45d450 100644
--- a/lib/librte_port/rte_port_ras.c
+++ b/lib/librte_port/rte_port_ras.c
@@ -144,7 +144,7 @@ rte_port_ring_writer_ras_create(void *params, int socket_id, int is_ipv4)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 
-	port->f_ras = (is_ipv4 == 0) ? process_ipv4 : process_ipv6;
+	port->f_ras = (is_ipv4 == 1) ? process_ipv4 : process_ipv6;
 
 	return port;
 }
@@ -182,7 +182,7 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 	/* Assume there is no ethernet header */
 	struct ipv4_hdr *pkt_hdr = rte_pktmbuf_mtod(pkt, struct ipv4_hdr *);
 
-	/* Get "Do not fragment" flag and fragment offset */
+	/* Get "More fragments" flag and fragment offset */
 	uint16_t frag_field = rte_be_to_cpu_16(pkt_hdr->fragment_offset);
 	uint16_t frag_offset = (uint16_t)(frag_field & IPV4_HDR_OFFSET_MASK);
 	uint16_t frag_flag = (uint16_t)(frag_field & IPV4_HDR_MF_FLAG);
@@ -195,6 +195,8 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv4_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(),
 				pkt_hdr);
@@ -224,6 +226,8 @@ process_ipv6(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr) + sizeof(*frag_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv6_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(), pkt_hdr,
 				frag_hdr);
-- 
1.7.9.5

[dpdk-dev] [PATCH v1 3/3] examples/ip_pipeline: add mp/mc and frag/ras swq

[Piotr Azarewicz]

Add integrated MP/MC and fragmentation/reassembly support to SWQs

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 examples/ip_pipeline/app.h          |   14 +++
 examples/ip_pipeline/config_check.c |   45 +++++++-
 examples/ip_pipeline/config_parse.c |  195 +++++++++++++++++++++++++++++++++--
 examples/ip_pipeline/init.c         |  165 ++++++++++++++++++++++++-----
 examples/ip_pipeline/main.c         |    4 +-
 examples/ip_pipeline/pipeline_be.h  |   18 ++++
 6 files changed, 402 insertions(+), 39 deletions(-)

diff --git a/examples/ip_pipeline/app.h b/examples/ip_pipeline/app.h
index 521e3a0..943466e 100644
--- a/examples/ip_pipeline/app.h
+++ b/examples/ip_pipeline/app.h
@@ -107,6 +107,14 @@ struct app_pktq_swq_params {
 	uint32_t dropless;
 	uint64_t n_retries;
 	uint32_t cpu_socket_id;
+	uint32_t ipv4_frag;
+	uint32_t ipv6_frag;
+	uint32_t ipv4_ras;
+	uint32_t ipv6_ras;
+	uint32_t mtu;
+	uint32_t metadata_size;
+	uint32_t mempool_direct_id;
+	uint32_t mempool_indirect_id;
 };
 
 #ifndef APP_FILE_NAME_SIZE
@@ -405,6 +413,10 @@ struct app_params {
 	char app_name[APP_APPNAME_SIZE];
 	const char *config_file;
 	const char *script_file;
+	const char *parser_file;
+	const char *output_file;
+	const char *preproc;
+	const char *preproc_args;
 	uint64_t port_mask;
 	uint32_t log_level;
 
@@ -880,6 +892,8 @@ int app_config_init(struct app_params *app);
 int app_config_args(struct app_params *app,
 	int argc, char **argv);
 
+int app_config_preproc(struct app_params *app);
+
 int app_config_parse(struct app_params *app,
 	const char *file_name);
 
diff --git a/examples/ip_pipeline/config_check.c b/examples/ip_pipeline/config_check.c
index 07f4c8b..8052bc4 100644
--- a/examples/ip_pipeline/config_check.c
+++ b/examples/ip_pipeline/config_check.c
@@ -33,6 +33,8 @@
 
 #include <stdio.h>
 
+#include <rte_ip.h>
+
 #include "app.h"
 
 static void
@@ -193,6 +195,7 @@ check_swqs(struct app_params *app)
 		struct app_pktq_swq_params *p = &app->swq_params[i];
 		uint32_t n_readers = app_swq_get_readers(app, p);
 		uint32_t n_writers = app_swq_get_writers(app, p);
+		uint32_t n_flags;
 
 		APP_CHECK((p->size > 0),
 			"%s size is 0\n", p->name);
@@ -217,14 +220,48 @@ check_swqs(struct app_params *app)
 		APP_CHECK((n_readers != 0),
 			"%s has no reader\n", p->name);
 
-		APP_CHECK((n_readers == 1),
-			"%s has more than one reader\n", p->name);
+		if (n_readers > 1)
+			APP_LOG(app, LOW, "%s has more than one reader", p->name);
 
 		APP_CHECK((n_writers != 0),
 			"%s has no writer\n", p->name);
 
-		APP_CHECK((n_writers == 1),
-			"%s has more than one writer\n", p->name);
+		if (n_writers > 1)
+			APP_LOG(app, LOW, "%s has more than one writer", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag + p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((n_flags < 2),
+			"%s has more than one fragmentation or reassembly mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_readers > 1) && (n_flags == 1))),
+			"%s has more than one reader when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_writers > 1) && (n_flags == 1))),
+			"%s has more than one writer when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		n_flags = p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((!((p->dropless == 1) && (n_flags == 1))),
+			"%s has dropless when reassembly mode enabled\n", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag;
+
+		if (n_flags == 1) {
+			uint16_t ip_hdr_size = (p->ipv4_frag) ? sizeof(struct ipv4_hdr) :
+				sizeof(struct ipv6_hdr);
+
+			APP_CHECK((p->mtu > ip_hdr_size),
+				"%s mtu size is smaller than ip header\n", p->name);
+
+			APP_CHECK((!((p->mtu - ip_hdr_size) % 8)),
+				"%s mtu size is incorrect\n", p->name);
+		}
 	}
 }
 
diff --git a/examples/ip_pipeline/config_parse.c b/examples/ip_pipeline/config_parse.c
index c9b78f9..a35bd3e 100644
--- a/examples/ip_pipeline/config_parse.c
+++ b/examples/ip_pipeline/config_parse.c
@@ -156,6 +156,14 @@ static const struct app_pktq_swq_params default_swq_params = {
 	.dropless = 0,
 	.n_retries = 0,
 	.cpu_socket_id = 0,
+	.ipv4_frag = 0,
+	.ipv6_frag = 0,
+	.ipv4_ras = 0,
+	.ipv6_ras = 0,
+	.mtu = 0,
+	.metadata_size = 0,
+	.mempool_direct_id = 0,
+	.mempool_indirect_id = 0,
 };
 
 struct app_pktq_tm_params default_tm_params = {
@@ -196,13 +204,15 @@ struct app_pipeline_params default_pipeline_params = {
 
 static const char app_usage[] =
 	"Usage: %s [-f CONFIG_FILE] [-s SCRIPT_FILE] -p PORT_MASK "
-	"[-l LOG_LEVEL]\n"
+	"[-l LOG_LEVEL] [--preproc PREPROCESSOR] [--preproc-args ARGS]\n"
 	"\n"
 	"Arguments:\n"
 	"\t-f CONFIG_FILE: Default config file is %s\n"
 	"\t-p PORT_MASK: Mask of NIC port IDs in hexadecimal format\n"
 	"\t-s SCRIPT_FILE: No CLI script file is run when not specified\n"
 	"\t-l LOG_LEVEL: 0 = NONE, 1 = HIGH PRIO (default), 2 = LOW PRIO\n"
+	"\t--preproc PREPROCESSOR: Configuration file pre-processor\n"
+	"\t--preproc-args ARGS: Arguments to be passed to pre-processor\n"
 	"\n";
 
 static void
@@ -1107,6 +1117,10 @@ parse_pipeline(struct app_params *app,
 			ret = parser_read_uint32(&param->timer_period,
 				ent->value);
 		else {
+			APP_CHECK((param->n_args < APP_MAX_PIPELINE_ARGS),
+				"CFG: [%s] out of memory",
+				section_name);
+
 			param->args_name[param->n_args] = strdup(ent->name);
 			param->args_value[param->n_args] = strdup(ent->value);
 
@@ -1397,6 +1411,7 @@ parse_swq(struct app_params *app,
 	struct app_pktq_swq_params *param;
 	struct rte_cfgfile_entry *entries;
 	int n_entries, ret, i;
+	unsigned frag_entries = 0;
 	ssize_t param_idx;
 
 	n_entries = rte_cfgfile_section_num_entries(cfg, section_name);
@@ -1438,6 +1453,71 @@ parse_swq(struct app_params *app,
 		else if (strcmp(ent->name, "cpu") == 0)
 			ret = parser_read_uint32(&param->cpu_socket_id,
 				ent->value);
+		else if (strcmp(ent->name, "ipv4_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1500;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1320;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv4_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "mtu") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->mtu,
+				ent->value);
+		} else if (strcmp(ent->name, "metadata_size") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->metadata_size,
+				ent->value);
+		} else if (strcmp(ent->name, "mempool_direct") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_direct_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		} else if (strcmp(ent->name, "mempool_indirect") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_indirect_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		}
 
 		APP_CHECK(ret != -ESRCH,
 			"CFG: [%s] entry '%s': unknown entry\n",
@@ -1450,6 +1530,13 @@ parse_swq(struct app_params *app,
 			ent->value);
 	}
 
+	if (frag_entries == 1) {
+		APP_CHECK(((param->ipv4_frag == 1) || (param->ipv6_frag == 1)),
+			"CFG: [%s] ipv4/ipv6 frag is off : unsupported entries on this"
+			" configuration\n",
+			section_name);
+	}
+
 	free(entries);
 }
 
@@ -1769,7 +1856,6 @@ parse_port_mask(struct app_params *app, uint64_t port_mask)
 int
 app_config_parse(struct app_params *app, const char *file_name)
 {
-	char config_file_out[APP_FILE_NAME_SIZE];
 	struct rte_cfgfile *cfg;
 	char **section_names;
 	int i, j, sect_count;
@@ -1851,11 +1937,7 @@ app_config_parse(struct app_params *app, const char *file_name)
 	APP_PARAM_COUNT(app->pipeline_params, app->n_pipelines);
 
 	/* Save configuration to output file */
-	snprintf(config_file_out,
-		APP_FILE_NAME_SIZE,
-		"%s.out",
-		app->config_file);
-	app_config_save(app, config_file_out);
+	app_config_save(app, app->output_file);
 
 	/* Load TM configuration files */
 	app_config_parse_tm(app);
@@ -2069,6 +2151,20 @@ save_swq_params(struct app_params *app, FILE *f)
 		fprintf(f, "%s = %s\n", "dropless", p->dropless ? "yes" : "no");
 		fprintf(f, "%s = %" PRIu64 "\n", "n_retries", p->n_retries);
 		fprintf(f, "%s = %" PRIu32 "\n", "cpu", p->cpu_socket_id);
+		fprintf(f, "%s = %s\n", "ipv4_frag", p->ipv4_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_frag", p->ipv6_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv4_ras", p->ipv4_ras ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_ras", p->ipv6_ras ? "yes" : "no");
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1)) {
+			fprintf(f, "%s = %" PRIu32 "\n", "mtu", p->mtu);
+			fprintf(f, "%s = %" PRIu32 "\n", "metadata_size", p->metadata_size);
+			fprintf(f, "%s = %s\n",
+				"mempool_direct",
+				app->mempool_params[p->mempool_direct_id].name);
+			fprintf(f, "%s = %s\n",
+				"mempool_indirect",
+				app->mempool_params[p->mempool_indirect_id].name);
+		}
 
 		fputc('\n', f);
 	}
@@ -2360,15 +2456,31 @@ app_config_init(struct app_params *app)
 	return 0;
 }
 
+static char *
+filenamedup(const char *filename, const char *suffix)
+{
+	char *s = malloc(strlen(filename) + strlen(suffix) + 1);
+
+	if (!s)
+		return NULL;
+
+	sprintf(s, "%s%s", filename, suffix);
+	return s;
+}
+
 int
 app_config_args(struct app_params *app, int argc, char **argv)
 {
-	int opt;
-	int option_index, f_present, s_present, p_present, l_present;
+	const char *optname;
+	int opt, option_index;
+	int f_present, s_present, p_present, l_present;
+	int preproc_present, preproc_params_present;
 	int scaned = 0;
 
 	static struct option lgopts[] = {
-		{NULL, 0, 0, 0}
+		{ "preproc", 1, 0, 0 },
+		{ "preproc-args", 1, 0, 0 },
+		{ NULL,  0, 0, 0 }
 	};
 
 	/* Copy application name */
@@ -2378,6 +2490,8 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	s_present = 0;
 	p_present = 0;
 	l_present = 0;
+	preproc_present = 0;
+	preproc_params_present = 0;
 
 	while ((opt = getopt_long(argc, argv, "f:s:p:l:", lgopts,
 			&option_index)) != EOF)
@@ -2443,6 +2557,32 @@ app_config_args(struct app_params *app, int argc, char **argv)
 
 			break;
 
+		case 0:
+			optname = lgopts[option_index].name;
+
+			if (strcmp(optname, "preproc") == 0) {
+				if (preproc_present)
+					rte_panic("Error: Preprocessor argument "
+						"is provided more than once\n");
+				preproc_present = 1;
+
+				app->preproc = strdup(optarg);
+				break;
+			}
+
+			if (strcmp(optname, "preproc-args") == 0) {
+				if (preproc_params_present)
+					rte_panic("Error: Preprocessor args "
+						"are provided more than once\n");
+				preproc_params_present = 1;
+
+				app->preproc_args = strdup(optarg);
+				break;
+			}
+
+			app_print_usage(argv[0]);
+			break;
+
 		default:
 			app_print_usage(argv[0]);
 		}
@@ -2453,5 +2593,40 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	if (!p_present)
 		rte_panic("Error: PORT_MASK is not provided\n");
 
+	/* Check dependencies between args */
+	if (preproc_params_present && (preproc_present == 0))
+		rte_panic("Error: Preprocessor args specified while "
+			"preprocessor is not defined\n");
+
+	app->parser_file = preproc_present ?
+		filenamedup(app->config_file, ".preproc") :
+		strdup(app->config_file);
+	app->output_file = filenamedup(app->config_file, ".out");
+
 	return 0;
 }
+
+int
+app_config_preproc(struct app_params *app)
+{
+	char buffer[256];
+	int status;
+
+	if (app->preproc == NULL)
+		return 0;
+
+	status = access(app->config_file, F_OK | R_OK);
+	APP_CHECK((status == 0), "Unable to open file %s", app->config_file);
+
+	snprintf(buffer, sizeof(buffer), "%s %s %s > %s",
+		app->preproc,
+		app->preproc_args ? app->preproc_args : "",
+		app->config_file,
+		app->parser_file);
+
+	status = system(buffer);
+	APP_CHECK((WIFEXITED(status) && (WEXITSTATUS(status) == 0)),
+		"Error while preprocessing file \"%s\"\n", app->config_file);
+
+	return status;
+}
diff --git a/examples/ip_pipeline/init.c b/examples/ip_pipeline/init.c
index 3f9c68d..46d044e 100644
--- a/examples/ip_pipeline/init.c
+++ b/examples/ip_pipeline/init.c
@@ -803,11 +803,43 @@ app_check_link(struct app_params *app)
 		rte_panic("Some links are DOWN\n");
 }
 
+static uint32_t
+is_any_swq_frag_or_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	for (i = 0; i < app->n_pktq_swq; i++) {
+		struct app_pktq_swq_params *p = &app->swq_params[i];
+
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1) ||
+			(p->ipv4_ras == 1) || (p->ipv6_ras == 1))
+			return 1;
+	}
+
+	return 0;
+}
+
+static void
+app_init_link_frag_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	if (is_any_swq_frag_or_ras(app)) {
+		for (i = 0; i < app->n_pktq_hwq_out; i++) {
+			struct app_pktq_hwq_out_params *p_txq = &app->hwq_out_params[i];
+
+			p_txq->conf.txq_flags &= ~ETH_TXQ_FLAGS_NOMULTSEGS;
+		}
+	}
+}
+
 static void
 app_init_link(struct app_params *app)
 {
 	uint32_t i;
 
+	app_init_link_frag_ras(app);
+
 	for (i = 0; i < app->n_links; i++) {
 		struct app_link_params *p_link = &app->link_params[i];
 		uint32_t link_id, n_hwq_in, n_hwq_out, j;
@@ -916,13 +948,19 @@ app_init_swq(struct app_params *app)
 
 	for (i = 0; i < app->n_pktq_swq; i++) {
 		struct app_pktq_swq_params *p = &app->swq_params[i];
+		unsigned flags = 0;
+
+		if (app_swq_get_readers(app, p) == 1)
+			flags |= RING_F_SC_DEQ;
+		if (app_swq_get_writers(app, p) == 1)
+			flags |= RING_F_SP_ENQ;
 
 		APP_LOG(app, HIGH, "Initializing %s...", p->name);
 		app->swq[i] = rte_ring_create(
 				p->name,
 				p->size,
 				p->cpu_socket_id,
-				RING_F_SP_ENQ | RING_F_SC_DEQ);
+				flags);
 
 		if (app->swq[i] == NULL)
 			rte_panic("%s init error\n", p->name);
@@ -1059,11 +1097,50 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_IN_SWQ:
-			out->type = PIPELINE_PORT_IN_RING_READER;
-			out->params.ring.ring = app->swq[in->id];
-			out->burst_size = app->swq_params[in->id].burst_read;
-			/* What about frag and ras ports? */
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_frag == 0) && (swq_params->ipv6_frag == 0)) {
+				if (app_swq_get_readers(app, swq_params) == 1) {
+					out->type = PIPELINE_PORT_IN_RING_READER;
+					out->params.ring.ring = app->swq[in->id];
+					out->burst_size = app->swq_params[in->id].burst_read;
+				} else {
+					out->type = PIPELINE_PORT_IN_RING_MULTI_READER;
+					out->params.ring_multi.ring = app->swq[in->id];
+					out->burst_size = swq_params->burst_read;
+				}
+			} else {
+				if (swq_params->ipv4_frag == 1) {
+					struct rte_port_ring_reader_ipv4_frag_params *params =
+						&out->params.ring_ipv4_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV4_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				} else {
+					struct rte_port_ring_reader_ipv6_frag_params *params =
+						&out->params.ring_ipv6_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV6_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				}
+			}
 			break;
+		}
 		case APP_PKTQ_IN_TM:
 			out->type = PIPELINE_PORT_IN_SCHED_READER;
 			out->params.sched.sched = app->tm[in->id];
@@ -1122,28 +1199,68 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_OUT_SWQ:
-			if (app->swq_params[in->id].dropless == 0) {
-				struct rte_port_ring_writer_params *params =
-					&out->params.ring;
-
-				out->type = PIPELINE_PORT_OUT_RING_WRITER;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_ras == 0) && (swq_params->ipv6_ras == 0)) {
+				if (app_swq_get_writers(app, swq_params) == 1) {
+					if (app->swq_params[in->id].dropless == 0) {
+						struct rte_port_ring_writer_params *params =
+							&out->params.ring;
+
+						out->type = PIPELINE_PORT_OUT_RING_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+					} else {
+						struct rte_port_ring_writer_nodrop_params
+							*params = &out->params.ring_nodrop;
+
+						out->type =
+							PIPELINE_PORT_OUT_RING_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+						params->n_retries =
+							app->swq_params[in->id].n_retries;
+					}
+				} else {
+					if (swq_params->dropless == 0) {
+						struct rte_port_ring_multi_writer_params *params =
+							&out->params.ring_multi;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+					} else {
+						struct rte_port_ring_multi_writer_nodrop_params
+							*params = &out->params.ring_multi_nodrop;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+						params->n_retries = swq_params->n_retries;
+					}
+				}
 			} else {
-				struct rte_port_ring_writer_nodrop_params
-					*params = &out->params.ring_nodrop;
-
-				out->type =
-					PIPELINE_PORT_OUT_RING_WRITER_NODROP;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
-				params->n_retries =
-					app->swq_params[in->id].n_retries;
+				if (swq_params->ipv4_ras == 1) {
+					struct rte_port_ring_writer_ipv4_ras_params *params =
+						&out->params.ring_ipv4_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				} else {
+					struct rte_port_ring_writer_ipv6_ras_params *params =
+						&out->params.ring_ipv6_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				}
 			}
-			/* What about frag and ras ports? */
 			break;
+		}
 		case APP_PKTQ_OUT_TM: {
 			struct rte_port_sched_writer_params *params =
 				&out->params.sched;
diff --git a/examples/ip_pipeline/main.c b/examples/ip_pipeline/main.c
index 862e2f2..4944dcf 100644
--- a/examples/ip_pipeline/main.c
+++ b/examples/ip_pipeline/main.c
@@ -45,7 +45,9 @@ main(int argc, char **argv)
 
 	app_config_args(&app, argc, argv);
 
-	app_config_parse(&app, app.config_file);
+	app_config_preproc(&app);
+
+	app_config_parse(&app, app.parser_file);
 
 	app_config_check(&app);
 
diff --git a/examples/ip_pipeline/pipeline_be.h b/examples/ip_pipeline/pipeline_be.h
index 51f1e4f..f7269c0 100644
--- a/examples/ip_pipeline/pipeline_be.h
+++ b/examples/ip_pipeline/pipeline_be.h
@@ -45,6 +45,7 @@
 enum pipeline_port_in_type {
 	PIPELINE_PORT_IN_ETHDEV_READER,
 	PIPELINE_PORT_IN_RING_READER,
+	PIPELINE_PORT_IN_RING_MULTI_READER,
 	PIPELINE_PORT_IN_RING_READER_IPV4_FRAG,
 	PIPELINE_PORT_IN_RING_READER_IPV6_FRAG,
 	PIPELINE_PORT_IN_SCHED_READER,
@@ -56,6 +57,7 @@ struct pipeline_port_in_params {
 	union {
 		struct rte_port_ethdev_reader_params ethdev;
 		struct rte_port_ring_reader_params ring;
+		struct rte_port_ring_multi_reader_params ring_multi;
 		struct rte_port_ring_reader_ipv4_frag_params ring_ipv4_frag;
 		struct rte_port_ring_reader_ipv6_frag_params ring_ipv6_frag;
 		struct rte_port_sched_reader_params sched;
@@ -72,6 +74,8 @@ pipeline_port_in_params_convert(struct pipeline_port_in_params  *p)
 		return (void *) &p->params.ethdev;
 	case PIPELINE_PORT_IN_RING_READER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return (void *) &p->params.ring_ipv4_frag;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -93,6 +97,8 @@ pipeline_port_in_params_get_ops(struct pipeline_port_in_params  *p)
 		return &rte_port_ethdev_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER:
 		return &rte_port_ring_reader_ops;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return &rte_port_ring_multi_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return &rte_port_ring_reader_ipv4_frag_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -110,7 +116,9 @@ enum pipeline_port_out_type {
 	PIPELINE_PORT_OUT_ETHDEV_WRITER,
 	PIPELINE_PORT_OUT_ETHDEV_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER,
 	PIPELINE_PORT_OUT_RING_WRITER_NODROP,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS,
 	PIPELINE_PORT_OUT_SCHED_WRITER,
@@ -123,7 +131,9 @@ struct pipeline_port_out_params {
 		struct rte_port_ethdev_writer_params ethdev;
 		struct rte_port_ethdev_writer_nodrop_params ethdev_nodrop;
 		struct rte_port_ring_writer_params ring;
+		struct rte_port_ring_multi_writer_params ring_multi;
 		struct rte_port_ring_writer_nodrop_params ring_nodrop;
+		struct rte_port_ring_multi_writer_nodrop_params ring_multi_nodrop;
 		struct rte_port_ring_writer_ipv4_ras_params ring_ipv4_ras;
 		struct rte_port_ring_writer_ipv6_ras_params ring_ipv6_ras;
 		struct rte_port_sched_writer_params sched;
@@ -140,8 +150,12 @@ pipeline_port_out_params_convert(struct pipeline_port_out_params  *p)
 		return (void *) &p->params.ethdev_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return (void *) &p->params.ring_nodrop;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return (void *) &p->params.ring_multi_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return (void *) &p->params.ring_ipv4_ras;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
@@ -164,8 +178,12 @@ pipeline_port_out_params_get_ops(struct pipeline_port_out_params  *p)
 		return &rte_port_ethdev_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return &rte_port_ring_writer_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return &rte_port_ring_multi_writer_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return &rte_port_ring_writer_nodrop_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return &rte_port_ring_multi_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return &rte_port_ring_writer_ipv4_ras_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
-- 
1.7.9.5

Re: [dpdk-dev] [PATCH v1 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[Dumitrescu, Cristian]

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Piotr Azarewicz
> Sent: Tuesday, September 15, 2015 4:07 PM
> To: dev@dpdk.org
> Subject: [dpdk-dev] [PATCH v1 0/3] ip_pipeline: add MP/MC and frag/ras
> support to SWQs
> 
> This patch set enhancement ip_pipeline application:
> - librte_port: add support for multi-producer/multi-consumer ring ports
> - librte_port: bug fixes for ring ports with IPv4/IPv6 reassembly support
> - ip_pipeline application: integrate MP/MC and fragmentation/reassembly
> support
>  to SWQs
> 

Acked-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Stephen Hemminger]

On Tue, 15 Sep 2015 15:06:33 +0200
Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:

> +static int
> +rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
> +{

Please break arguments on line so that line length is not over 80 characters.

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Stephen Hemminger]

On Tue, 15 Sep 2015 15:06:33 +0200
Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:

> +static inline void
> +send_burst_mp(struct rte_port_ring_writer *p)
> +{

compiler will inline static functions anyway. No need to add inline qualifier

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Stephen Hemminger]

On Tue, 15 Sep 2015 15:06:33 +0200
Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:

> +		/*
> +		 * If we didnt manage to send all packets in single burst, move

Checkpatch complains:
WARNING: 'didnt' may be misspelled - perhaps 'didn't'?
#413: FILE: lib/librte_port/rte_port_ring.c:827:
+		 * If we didnt manage to send all packets in single burst, move

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Dumitrescu, Cristian]

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Stephen
> Hemminger
> Sent: Tuesday, September 22, 2015 1:36 AM
> To: Azarewicz, PiotrX T
> Cc: dev@dpdk.org
> Subject: Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports
> 
> On Tue, 15 Sep 2015 15:06:33 +0200
> Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:
> 
> > +static inline void
> > +send_burst_mp(struct rte_port_ring_writer *p)
> > +{
> 
> compiler will inline static functions anyway. No need to add inline qualifier

Hi Stephen,

Using 'static inline' seems to be the standard practice in DPDK and a good practice as well.

DPDK> grep 'static inline' `find -name '*.[hc]'` | wc -l
1700

Regards,
Cristian

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Dumitrescu, Cristian]

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Stephen
> Hemminger
> Sent: Tuesday, September 22, 2015 1:37 AM
> To: Azarewicz, PiotrX T
> Cc: dev@dpdk.org
> Subject: Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports
> 
> On Tue, 15 Sep 2015 15:06:33 +0200
> Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:
> 
> > +		/*
> > +		 * If we didnt manage to send all packets in single burst,
> move
> 
> Checkpatch complains:
> WARNING: 'didnt' may be misspelled - perhaps 'didn't'?
> #413: FILE: lib/librte_port/rte_port_ring.c:827:
> +		 * If we didnt manage to send all packets in single burst,
> move

Thanks for the catch, Stephen.

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Dumitrescu, Cristian]

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Stephen
> Hemminger
> Sent: Tuesday, September 22, 2015 1:35 AM
> To: Azarewicz, PiotrX T
> Cc: dev@dpdk.org
> Subject: Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports
> 
> On Tue, 15 Sep 2015 15:06:33 +0200
> Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:
> 
> > +static int
> > +rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts,
> uint32_t n_pkts)
> > +{
> 
> Please break arguments on line so that line length is not over 80 characters.

Thanks, Steve.

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Thomas Monjalon]

2015-09-22 11:34, Dumitrescu, Cristian:
> 
> > -----Original Message-----
> > From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Stephen
> > Hemminger
> > Sent: Tuesday, September 22, 2015 1:36 AM
> > To: Azarewicz, PiotrX T
> > Cc: dev@dpdk.org
> > Subject: Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports
> > 
> > On Tue, 15 Sep 2015 15:06:33 +0200
> > Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:
> > 
> > > +static inline void
> > > +send_burst_mp(struct rte_port_ring_writer *p)
> > > +{
> > 
> > compiler will inline static functions anyway. No need to add inline qualifier
> 
> Hi Stephen,
> 
> Using 'static inline' seems to be the standard practice in DPDK and a good practice as well.

Why do you think it is a good practice?
Forced inlining can be a random optimization having negative effects.

> DPDK> grep 'static inline' `find -name '*.[hc]'` | wc -l
> 1700

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Stephen Hemminger]

On Tue, 22 Sep 2015 16:23:51 +0200
Thomas Monjalon <thomas.monjalon@6wind.com> wrote:

> 2015-09-22 11:34, Dumitrescu, Cristian:
> > 
> > > -----Original Message-----
> > > From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Stephen
> > > Hemminger
> > > Sent: Tuesday, September 22, 2015 1:36 AM
> > > To: Azarewicz, PiotrX T
> > > Cc: dev@dpdk.org
> > > Subject: Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports
> > > 
> > > On Tue, 15 Sep 2015 15:06:33 +0200
> > > Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:
> > > 
> > > > +static inline void
> > > > +send_burst_mp(struct rte_port_ring_writer *p)
> > > > +{
> > > 
> > > compiler will inline static functions anyway. No need to add inline qualifier
> > 
> > Hi Stephen,
> > 
> > Using 'static inline' seems to be the standard practice in DPDK and a good practice as well.
> 
> Why do you think it is a good practice?
> Forced inlining can be a random optimization having negative effects.

Agreed. Modern compilers make good decisions.

Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports

[Dumitrescu, Cristian]

> -----Original Message-----
> From: Thomas Monjalon [mailto:thomas.monjalon@6wind.com]
> Sent: Tuesday, September 22, 2015 5:24 PM
> To: Dumitrescu, Cristian
> Cc: dev@dpdk.org; Stephen Hemminger; Azarewicz, PiotrX T
> Subject: Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports
> 
> 2015-09-22 11:34, Dumitrescu, Cristian:
> >
> > > -----Original Message-----
> > > From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Stephen
> > > Hemminger
> > > Sent: Tuesday, September 22, 2015 1:36 AM
> > > To: Azarewicz, PiotrX T
> > > Cc: dev@dpdk.org
> > > Subject: Re: [dpdk-dev] [PATCH v1 1/3] port: add mp/mc ring ports
> > >
> > > On Tue, 15 Sep 2015 15:06:33 +0200
> > > Piotr Azarewicz <piotrx.t.azarewicz@intel.com> wrote:
> > >
> > > > +static inline void
> > > > +send_burst_mp(struct rte_port_ring_writer *p)
> > > > +{
> > >
> > > compiler will inline static functions anyway. No need to add inline qualifier
> >
> > Hi Stephen,
> >
> > Using 'static inline' seems to be the standard practice in DPDK and a good
> practice as well.
> 
> Why do you think it is a good practice?
> Forced inlining can be a random optimization having negative effects.
> 

What I meant is this: when users want to make sure their code gets inlined by the compiler, it is better to explicitly state this by using the mechanisms provided by the C compiler (C keyword "inline" and compiler pragmas like "always inline") rather than hope that compiler is going to do this anyway. I have been burned in the past by compilers not inlining code even when explicitly stated, so I am a quite sceptical about compilers doing it proactively.

Your point is slightly different: why use code inlining at all? IMHO this discussion is outside the scope of this patch and should be conducted as a separate debate. Please feel free to start it as a separate thread if you deem necessary. As said, there are already 1700 instances of "static inline" in DPDK, as well as lots of "always inline".

In the context of this debate (outside the scope of this patch), my quick input is:  compilers are typically good to optimize code at the function level rather than cross-functions, so having more code in the same function allows the compiler to do a better job at code optimization. I am not a compiler expert, so my views could simply be biased by my past experience.

> > DPDK> grep 'static inline' `find -name '*.[hc]'` | wc -l
> > 1700

[dpdk-dev] [PATCH v2 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[Piotr Azarewicz]

This patch set enhancement ip_pipeline application:
- librte_port: add support for multi-producer/multi-consumer ring ports
- librte_port: bug fixes for ring ports with IPv4/IPv6 reassembly support
- ip_pipeline application: integrate MP/MC and fragmentation/reassembly support
 to SWQs

v2 changes:
- rte_port_ring:
	- fixed checkpatch errors
	- interlace the implementation of multi into the implementation of single
	- reduced the amount of code duplication

Piotr Azarewicz (3):
  port: add mp/mc ring ports
  port: fix ras ring ports
  examples/ip_pipeline: add mp/mc and frag/ras swq

 examples/ip_pipeline/app.h          |   14 ++
 examples/ip_pipeline/config_check.c |   45 ++++-
 examples/ip_pipeline/config_parse.c |  195 ++++++++++++++++++++--
 examples/ip_pipeline/init.c         |  165 ++++++++++++++++---
 examples/ip_pipeline/main.c         |    4 +-
 examples/ip_pipeline/pipeline_be.h  |   18 ++
 lib/librte_port/rte_port_ras.c      |    8 +-
 lib/librte_port/rte_port_ring.c     |  311 ++++++++++++++++++++++++++++++++---
 lib/librte_port/rte_port_ring.h     |   35 +++-
 9 files changed, 724 insertions(+), 71 deletions(-)

-- 
1.7.9.5

[dpdk-dev] [PATCH v2 1/3] port: add mp/mc ring ports

[Piotr Azarewicz]

ring_multi_reader input port (on top of multi consumer rte_ring)
ring_multi_writer output port (on top of multi producer rte_ring)

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_ring.c |  311 +++++++++++++++++++++++++++++++++++----
 lib/librte_port/rte_port_ring.h |   35 ++++-
 2 files changed, 316 insertions(+), 30 deletions(-)

diff --git a/lib/librte_port/rte_port_ring.c b/lib/librte_port/rte_port_ring.c
index 9461c05..755dfc1 100644
--- a/lib/librte_port/rte_port_ring.c
+++ b/lib/librte_port/rte_port_ring.c
@@ -63,15 +63,19 @@ struct rte_port_ring_reader {
 };
 
 static void *
-rte_port_ring_reader_create(void *params, int socket_id)
+rte_port_ring_reader_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_reader_params *conf =
 			(struct rte_port_ring_reader_params *) params;
 	struct rte_port_ring_reader *port;
 
 	/* Check input parameters */
-	if (conf == NULL) {
-		RTE_LOG(ERR, PORT, "%s: params is NULL\n", __func__);
+	if ((conf == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->cons.sc_dequeue && is_multi) ||
+		(!(conf->ring->cons.sc_dequeue) && !is_multi)) {
+		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
 	}
 
@@ -89,6 +93,18 @@ rte_port_ring_reader_create(void *params, int socket_id)
 	return port;
 }
 
+static void *
+rte_port_ring_reader_create(void *params, int socket_id)
+{
+	return rte_port_ring_reader_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_reader_create(void *params, int socket_id)
+{
+	return rte_port_ring_reader_create_internal(params, socket_id, 1);
+}
+
 static int
 rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
 {
@@ -102,6 +118,19 @@ rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
 }
 
 static int
+rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts,
+	uint32_t n_pkts)
+{
+	struct rte_port_ring_reader *p = (struct rte_port_ring_reader *) port;
+	uint32_t nb_rx;
+
+	nb_rx = rte_ring_mc_dequeue_burst(p->ring, (void **) pkts, n_pkts);
+	RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx);
+
+	return nb_rx;
+}
+
+static int
 rte_port_ring_reader_free(void *port)
 {
 	if (port == NULL) {
@@ -155,10 +184,12 @@ struct rte_port_ring_writer {
 	uint32_t tx_burst_sz;
 	uint32_t tx_buf_count;
 	uint64_t bsz_mask;
+	uint32_t is_multi;
 };
 
 static void *
-rte_port_ring_writer_create(void *params, int socket_id)
+rte_port_ring_writer_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_writer_params *conf =
 			(struct rte_port_ring_writer_params *) params;
@@ -166,7 +197,9 @@ rte_port_ring_writer_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue && is_multi) ||
+		(!(conf->ring->prod.sp_enqueue) && !is_multi) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -185,10 +218,23 @@ rte_port_ring_writer_create(void *params, int socket_id)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+	port->is_multi = is_multi;
 
 	return port;
 }
 
+static void *
+rte_port_ring_writer_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_writer_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_create_internal(params, socket_id, 1);
+}
+
 static inline void
 send_burst(struct rte_port_ring_writer *p)
 {
@@ -204,6 +250,21 @@ send_burst(struct rte_port_ring_writer *p)
 	p->tx_buf_count = 0;
 }
 
+static inline void
+send_burst_mp(struct rte_port_ring_writer *p)
+{
+	uint32_t nb_tx;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+			p->tx_buf_count);
+
+	RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
 static int
 rte_port_ring_writer_tx(void *port, struct rte_mbuf *pkt)
 {
@@ -218,9 +279,23 @@ rte_port_ring_writer_tx(void *port, struct rte_mbuf *pkt)
 }
 
 static int
-rte_port_ring_writer_tx_bulk(void *port,
+rte_port_ring_multi_writer_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static inline int __attribute__((always_inline))
+rte_port_ring_writer_tx_bulk_internal(void *port,
 		struct rte_mbuf **pkts,
-		uint64_t pkts_mask)
+		uint64_t pkts_mask,
+		uint32_t is_multi)
 {
 	struct rte_port_ring_writer *p =
 		(struct rte_port_ring_writer *) port;
@@ -234,11 +309,20 @@ rte_port_ring_writer_tx_bulk(void *port,
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t n_pkts_ok;
 
-		if (tx_buf_count)
-			send_burst(p);
+		if (tx_buf_count) {
+			if (is_multi)
+				send_burst_mp(p);
+			else
+				send_burst(p);
+		}
 
 		RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, n_pkts);
-		n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		if (is_multi)
+			n_pkts_ok = rte_ring_mp_enqueue_burst(p->ring, (void **)pkts,
+				n_pkts);
+		else
+			n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts,
+				n_pkts);
 
 		RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, n_pkts - n_pkts_ok);
 		for ( ; n_pkts_ok < n_pkts; n_pkts_ok++) {
@@ -258,14 +342,34 @@ rte_port_ring_writer_tx_bulk(void *port,
 		}
 
 		p->tx_buf_count = tx_buf_count;
-		if (tx_buf_count >= p->tx_burst_sz)
-			send_burst(p);
+		if (tx_buf_count >= p->tx_burst_sz) {
+			if (is_multi)
+				send_burst_mp(p);
+			else
+				send_burst(p);
+		}
 	}
 
 	return 0;
 }
 
 static int
+rte_port_ring_writer_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 0);
+}
+
+static int
+rte_port_ring_multi_writer_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 1);
+}
+
+static int
 rte_port_ring_writer_flush(void *port)
 {
 	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
@@ -277,14 +381,31 @@ rte_port_ring_writer_flush(void *port)
 }
 
 static int
+rte_port_ring_multi_writer_flush(void *port)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static int
 rte_port_ring_writer_free(void *port)
 {
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
 	if (port == NULL) {
 		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
 		return -EINVAL;
 	}
 
-	rte_port_ring_writer_flush(port);
+	if (p->is_multi)
+		rte_port_ring_multi_writer_flush(port);
+	else
+		rte_port_ring_writer_flush(port);
+
 	rte_free(port);
 
 	return 0;
@@ -332,10 +453,12 @@ struct rte_port_ring_writer_nodrop {
 	uint32_t tx_buf_count;
 	uint64_t bsz_mask;
 	uint64_t n_retries;
+	uint32_t is_multi;
 };
 
 static void *
-rte_port_ring_writer_nodrop_create(void *params, int socket_id)
+rte_port_ring_writer_nodrop_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_writer_nodrop_params *conf =
 			(struct rte_port_ring_writer_nodrop_params *) params;
@@ -343,7 +466,9 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue && is_multi) ||
+		(!(conf->ring->prod.sp_enqueue) && !is_multi) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -362,6 +487,7 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+	port->is_multi = is_multi;
 
 	/*
 	 * When n_retries is 0 it means that we should wait for every packet to
@@ -373,6 +499,18 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 	return port;
 }
 
+static void *
+rte_port_ring_writer_nodrop_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_writer_nodrop_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 1);
+}
+
 static inline void
 send_burst_nodrop(struct rte_port_ring_writer_nodrop *p)
 {
@@ -402,6 +540,35 @@ send_burst_nodrop(struct rte_port_ring_writer_nodrop *p)
 	p->tx_buf_count = 0;
 }
 
+static inline void
+send_burst_mp_nodrop(struct rte_port_ring_writer_nodrop *p)
+{
+	uint32_t nb_tx = 0, i;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+				p->tx_buf_count);
+
+	/* We sent all the packets in a first try */
+	if (nb_tx >= p->tx_buf_count)
+		return;
+
+	for (i = 0; i < p->n_retries; i++) {
+		nb_tx += rte_ring_mp_enqueue_burst(p->ring,
+				(void **) (p->tx_buf + nb_tx), p->tx_buf_count - nb_tx);
+
+		/* We sent all the packets in more than one try */
+		if (nb_tx >= p->tx_buf_count)
+			return;
+	}
+
+	/* We didn't send the packets in maximum allowed attempts */
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
 static int
 rte_port_ring_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
 {
@@ -417,9 +584,24 @@ rte_port_ring_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
 }
 
 static int
-rte_port_ring_writer_nodrop_tx_bulk(void *port,
+rte_port_ring_multi_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static inline int __attribute__((always_inline))
+rte_port_ring_writer_nodrop_tx_bulk_internal(void *port,
 		struct rte_mbuf **pkts,
-		uint64_t pkts_mask)
+		uint64_t pkts_mask,
+		uint32_t is_multi)
 {
 	struct rte_port_ring_writer_nodrop *p =
 		(struct rte_port_ring_writer_nodrop *) port;
@@ -433,24 +615,37 @@ rte_port_ring_writer_nodrop_tx_bulk(void *port,
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t n_pkts_ok;
 
-		if (tx_buf_count)
-			send_burst_nodrop(p);
+		if (tx_buf_count) {
+			if (is_multi)
+				send_burst_mp_nodrop(p);
+			else
+				send_burst_nodrop(p);
+		}
 
 		RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, n_pkts);
-		n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		if (is_multi)
+			n_pkts_ok =
+				rte_ring_mp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		else
+			n_pkts_ok =
+				rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
 
 		if (n_pkts_ok >= n_pkts)
 			return 0;
 
 		/*
-		 * If we didnt manage to send all packets in single burst, move
+		 * If we didn't manage to send all packets in single burst, move
 		 * remaining packets to the buffer and call send burst.
 		 */
 		for (; n_pkts_ok < n_pkts; n_pkts_ok++) {
 			struct rte_mbuf *pkt = pkts[n_pkts_ok];
+
 			p->tx_buf[p->tx_buf_count++] = pkt;
 		}
-		send_burst_nodrop(p);
+		if (is_multi)
+			send_burst_mp_nodrop(p);
+		else
+			send_burst_nodrop(p);
 	} else {
 		for ( ; pkts_mask; ) {
 			uint32_t pkt_index = __builtin_ctzll(pkts_mask);
@@ -463,14 +658,36 @@ rte_port_ring_writer_nodrop_tx_bulk(void *port,
 		}
 
 		p->tx_buf_count = tx_buf_count;
-		if (tx_buf_count >= p->tx_burst_sz)
-			send_burst_nodrop(p);
+		if (tx_buf_count >= p->tx_burst_sz) {
+			if (is_multi)
+				send_burst_mp_nodrop(p);
+			else
+				send_burst_nodrop(p);
+		}
 	}
 
 	return 0;
 }
 
 static int
+rte_port_ring_writer_nodrop_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return
+		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 0);
+}
+
+static int
+rte_port_ring_multi_writer_nodrop_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return
+		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 1);
+}
+
+static int
 rte_port_ring_writer_nodrop_flush(void *port)
 {
 	struct rte_port_ring_writer_nodrop *p =
@@ -483,14 +700,33 @@ rte_port_ring_writer_nodrop_flush(void *port)
 }
 
 static int
+rte_port_ring_multi_writer_nodrop_flush(void *port)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static int
 rte_port_ring_writer_nodrop_free(void *port)
 {
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
 	if (port == NULL) {
 		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
 		return -EINVAL;
 	}
 
-	rte_port_ring_writer_nodrop_flush(port);
+	if (p->is_multi)
+		rte_port_ring_multi_writer_nodrop_flush(port);
+	else
+		rte_port_ring_writer_nodrop_flush(port);
+
 	rte_free(port);
 
 	return 0;
@@ -539,3 +775,28 @@ struct rte_port_out_ops rte_port_ring_writer_nodrop_ops = {
 	.f_flush = rte_port_ring_writer_nodrop_flush,
 	.f_stats = rte_port_ring_writer_nodrop_stats_read,
 };
+
+struct rte_port_in_ops rte_port_ring_multi_reader_ops = {
+	.f_create = rte_port_ring_multi_reader_create,
+	.f_free = rte_port_ring_reader_free,
+	.f_rx = rte_port_ring_multi_reader_rx,
+	.f_stats = rte_port_ring_reader_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_ops = {
+	.f_create = rte_port_ring_multi_writer_create,
+	.f_free = rte_port_ring_writer_free,
+	.f_tx = rte_port_ring_multi_writer_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_flush,
+	.f_stats = rte_port_ring_writer_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops = {
+	.f_create = rte_port_ring_multi_writer_nodrop_create,
+	.f_free = rte_port_ring_writer_nodrop_free,
+	.f_tx = rte_port_ring_multi_writer_nodrop_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_nodrop_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_nodrop_flush,
+	.f_stats = rte_port_ring_writer_nodrop_stats_read,
+};
diff --git a/lib/librte_port/rte_port_ring.h b/lib/librte_port/rte_port_ring.h
index 89a219b..de377d2 100644
--- a/lib/librte_port/rte_port_ring.h
+++ b/lib/librte_port/rte_port_ring.h
@@ -42,8 +42,14 @@ extern "C" {
  * @file
  * RTE Port Ring
  *
- * ring_reader: input port built on top of pre-initialized single consumer ring
- * ring_writer: output port built on top of pre-initialized single producer ring
+ * ring_reader:
+ *      input port built on top of pre-initialized single consumer ring
+ * ring_writer:
+ *      output port built on top of pre-initialized single producer ring
+ * ring_multi_reader:
+ *      input port built on top of pre-initialized multi consumers ring
+ * ring_multi_writer:
+ *      output port built on top of pre-initialized multi producers ring
  *
  ***/
 
@@ -55,7 +61,7 @@ extern "C" {
 
 /** ring_reader port parameters */
 struct rte_port_ring_reader_params {
-	/** Underlying single consumer ring that has to be pre-initialized */
+	/** Underlying consumer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 };
 
@@ -64,7 +70,7 @@ extern struct rte_port_in_ops rte_port_ring_reader_ops;
 
 /** ring_writer port parameters */
 struct rte_port_ring_writer_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -77,7 +83,7 @@ extern struct rte_port_out_ops rte_port_ring_writer_ops;
 
 /** ring_writer_nodrop port parameters */
 struct rte_port_ring_writer_nodrop_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -91,6 +97,25 @@ struct rte_port_ring_writer_nodrop_params {
 /** ring_writer_nodrop port operations */
 extern struct rte_port_out_ops rte_port_ring_writer_nodrop_ops;
 
+/** ring_multi_reader port parameters */
+#define rte_port_ring_multi_reader_params rte_port_ring_reader_params
+
+/** ring_multi_reader port operations */
+extern struct rte_port_in_ops rte_port_ring_multi_reader_ops;
+
+/** ring_multi_writer port parameters */
+#define rte_port_ring_multi_writer_params rte_port_ring_writer_params
+
+/** ring_multi_writer port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_ops;
+
+/** ring_multi_writer_nodrop port parameters */
+#define rte_port_ring_multi_writer_nodrop_params \
+	rte_port_ring_writer_nodrop_params
+
+/** ring_multi_writer_nodrop port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops;
+
 #ifdef __cplusplus
 }
 #endif
-- 
1.7.9.5

[dpdk-dev] [PATCH v2 2/3] port: fix ras ring ports

[Piotr Azarewicz]

Bug fixes for ring ports with IPv4/IPv6 reassembly support.
Previous implementation can't work properly due to incorrect choosing
process function.
Also, assuming that, when processing ip packet, ip header is know we can
set l3_len parameter here.

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_ras.c |    8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/lib/librte_port/rte_port_ras.c b/lib/librte_port/rte_port_ras.c
index 6bd0f8c..e45d450 100644
--- a/lib/librte_port/rte_port_ras.c
+++ b/lib/librte_port/rte_port_ras.c
@@ -144,7 +144,7 @@ rte_port_ring_writer_ras_create(void *params, int socket_id, int is_ipv4)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 
-	port->f_ras = (is_ipv4 == 0) ? process_ipv4 : process_ipv6;
+	port->f_ras = (is_ipv4 == 1) ? process_ipv4 : process_ipv6;
 
 	return port;
 }
@@ -182,7 +182,7 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 	/* Assume there is no ethernet header */
 	struct ipv4_hdr *pkt_hdr = rte_pktmbuf_mtod(pkt, struct ipv4_hdr *);
 
-	/* Get "Do not fragment" flag and fragment offset */
+	/* Get "More fragments" flag and fragment offset */
 	uint16_t frag_field = rte_be_to_cpu_16(pkt_hdr->fragment_offset);
 	uint16_t frag_offset = (uint16_t)(frag_field & IPV4_HDR_OFFSET_MASK);
 	uint16_t frag_flag = (uint16_t)(frag_field & IPV4_HDR_MF_FLAG);
@@ -195,6 +195,8 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv4_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(),
 				pkt_hdr);
@@ -224,6 +226,8 @@ process_ipv6(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr) + sizeof(*frag_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv6_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(), pkt_hdr,
 				frag_hdr);
-- 
1.7.9.5

[dpdk-dev] [PATCH v2 3/3] examples/ip_pipeline: add mp/mc and frag/ras swq

[Piotr Azarewicz]

Add integrated MP/MC and fragmentation/reassembly support to SWQs

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 examples/ip_pipeline/app.h          |   14 +++
 examples/ip_pipeline/config_check.c |   45 +++++++-
 examples/ip_pipeline/config_parse.c |  195 +++++++++++++++++++++++++++++++++--
 examples/ip_pipeline/init.c         |  165 ++++++++++++++++++++++++-----
 examples/ip_pipeline/main.c         |    4 +-
 examples/ip_pipeline/pipeline_be.h  |   18 ++++
 6 files changed, 402 insertions(+), 39 deletions(-)

diff --git a/examples/ip_pipeline/app.h b/examples/ip_pipeline/app.h
index 521e3a0..943466e 100644
--- a/examples/ip_pipeline/app.h
+++ b/examples/ip_pipeline/app.h
@@ -107,6 +107,14 @@ struct app_pktq_swq_params {
 	uint32_t dropless;
 	uint64_t n_retries;
 	uint32_t cpu_socket_id;
+	uint32_t ipv4_frag;
+	uint32_t ipv6_frag;
+	uint32_t ipv4_ras;
+	uint32_t ipv6_ras;
+	uint32_t mtu;
+	uint32_t metadata_size;
+	uint32_t mempool_direct_id;
+	uint32_t mempool_indirect_id;
 };
 
 #ifndef APP_FILE_NAME_SIZE
@@ -405,6 +413,10 @@ struct app_params {
 	char app_name[APP_APPNAME_SIZE];
 	const char *config_file;
 	const char *script_file;
+	const char *parser_file;
+	const char *output_file;
+	const char *preproc;
+	const char *preproc_args;
 	uint64_t port_mask;
 	uint32_t log_level;
 
@@ -880,6 +892,8 @@ int app_config_init(struct app_params *app);
 int app_config_args(struct app_params *app,
 	int argc, char **argv);
 
+int app_config_preproc(struct app_params *app);
+
 int app_config_parse(struct app_params *app,
 	const char *file_name);
 
diff --git a/examples/ip_pipeline/config_check.c b/examples/ip_pipeline/config_check.c
index 07f4c8b..8052bc4 100644
--- a/examples/ip_pipeline/config_check.c
+++ b/examples/ip_pipeline/config_check.c
@@ -33,6 +33,8 @@
 
 #include <stdio.h>
 
+#include <rte_ip.h>
+
 #include "app.h"
 
 static void
@@ -193,6 +195,7 @@ check_swqs(struct app_params *app)
 		struct app_pktq_swq_params *p = &app->swq_params[i];
 		uint32_t n_readers = app_swq_get_readers(app, p);
 		uint32_t n_writers = app_swq_get_writers(app, p);
+		uint32_t n_flags;
 
 		APP_CHECK((p->size > 0),
 			"%s size is 0\n", p->name);
@@ -217,14 +220,48 @@ check_swqs(struct app_params *app)
 		APP_CHECK((n_readers != 0),
 			"%s has no reader\n", p->name);
 
-		APP_CHECK((n_readers == 1),
-			"%s has more than one reader\n", p->name);
+		if (n_readers > 1)
+			APP_LOG(app, LOW, "%s has more than one reader", p->name);
 
 		APP_CHECK((n_writers != 0),
 			"%s has no writer\n", p->name);
 
-		APP_CHECK((n_writers == 1),
-			"%s has more than one writer\n", p->name);
+		if (n_writers > 1)
+			APP_LOG(app, LOW, "%s has more than one writer", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag + p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((n_flags < 2),
+			"%s has more than one fragmentation or reassembly mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_readers > 1) && (n_flags == 1))),
+			"%s has more than one reader when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_writers > 1) && (n_flags == 1))),
+			"%s has more than one writer when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		n_flags = p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((!((p->dropless == 1) && (n_flags == 1))),
+			"%s has dropless when reassembly mode enabled\n", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag;
+
+		if (n_flags == 1) {
+			uint16_t ip_hdr_size = (p->ipv4_frag) ? sizeof(struct ipv4_hdr) :
+				sizeof(struct ipv6_hdr);
+
+			APP_CHECK((p->mtu > ip_hdr_size),
+				"%s mtu size is smaller than ip header\n", p->name);
+
+			APP_CHECK((!((p->mtu - ip_hdr_size) % 8)),
+				"%s mtu size is incorrect\n", p->name);
+		}
 	}
 }
 
diff --git a/examples/ip_pipeline/config_parse.c b/examples/ip_pipeline/config_parse.c
index c9b78f9..a35bd3e 100644
--- a/examples/ip_pipeline/config_parse.c
+++ b/examples/ip_pipeline/config_parse.c
@@ -156,6 +156,14 @@ static const struct app_pktq_swq_params default_swq_params = {
 	.dropless = 0,
 	.n_retries = 0,
 	.cpu_socket_id = 0,
+	.ipv4_frag = 0,
+	.ipv6_frag = 0,
+	.ipv4_ras = 0,
+	.ipv6_ras = 0,
+	.mtu = 0,
+	.metadata_size = 0,
+	.mempool_direct_id = 0,
+	.mempool_indirect_id = 0,
 };
 
 struct app_pktq_tm_params default_tm_params = {
@@ -196,13 +204,15 @@ struct app_pipeline_params default_pipeline_params = {
 
 static const char app_usage[] =
 	"Usage: %s [-f CONFIG_FILE] [-s SCRIPT_FILE] -p PORT_MASK "
-	"[-l LOG_LEVEL]\n"
+	"[-l LOG_LEVEL] [--preproc PREPROCESSOR] [--preproc-args ARGS]\n"
 	"\n"
 	"Arguments:\n"
 	"\t-f CONFIG_FILE: Default config file is %s\n"
 	"\t-p PORT_MASK: Mask of NIC port IDs in hexadecimal format\n"
 	"\t-s SCRIPT_FILE: No CLI script file is run when not specified\n"
 	"\t-l LOG_LEVEL: 0 = NONE, 1 = HIGH PRIO (default), 2 = LOW PRIO\n"
+	"\t--preproc PREPROCESSOR: Configuration file pre-processor\n"
+	"\t--preproc-args ARGS: Arguments to be passed to pre-processor\n"
 	"\n";
 
 static void
@@ -1107,6 +1117,10 @@ parse_pipeline(struct app_params *app,
 			ret = parser_read_uint32(&param->timer_period,
 				ent->value);
 		else {
+			APP_CHECK((param->n_args < APP_MAX_PIPELINE_ARGS),
+				"CFG: [%s] out of memory",
+				section_name);
+
 			param->args_name[param->n_args] = strdup(ent->name);
 			param->args_value[param->n_args] = strdup(ent->value);
 
@@ -1397,6 +1411,7 @@ parse_swq(struct app_params *app,
 	struct app_pktq_swq_params *param;
 	struct rte_cfgfile_entry *entries;
 	int n_entries, ret, i;
+	unsigned frag_entries = 0;
 	ssize_t param_idx;
 
 	n_entries = rte_cfgfile_section_num_entries(cfg, section_name);
@@ -1438,6 +1453,71 @@ parse_swq(struct app_params *app,
 		else if (strcmp(ent->name, "cpu") == 0)
 			ret = parser_read_uint32(&param->cpu_socket_id,
 				ent->value);
+		else if (strcmp(ent->name, "ipv4_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1500;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1320;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv4_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "mtu") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->mtu,
+				ent->value);
+		} else if (strcmp(ent->name, "metadata_size") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->metadata_size,
+				ent->value);
+		} else if (strcmp(ent->name, "mempool_direct") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_direct_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		} else if (strcmp(ent->name, "mempool_indirect") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_indirect_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		}
 
 		APP_CHECK(ret != -ESRCH,
 			"CFG: [%s] entry '%s': unknown entry\n",
@@ -1450,6 +1530,13 @@ parse_swq(struct app_params *app,
 			ent->value);
 	}
 
+	if (frag_entries == 1) {
+		APP_CHECK(((param->ipv4_frag == 1) || (param->ipv6_frag == 1)),
+			"CFG: [%s] ipv4/ipv6 frag is off : unsupported entries on this"
+			" configuration\n",
+			section_name);
+	}
+
 	free(entries);
 }
 
@@ -1769,7 +1856,6 @@ parse_port_mask(struct app_params *app, uint64_t port_mask)
 int
 app_config_parse(struct app_params *app, const char *file_name)
 {
-	char config_file_out[APP_FILE_NAME_SIZE];
 	struct rte_cfgfile *cfg;
 	char **section_names;
 	int i, j, sect_count;
@@ -1851,11 +1937,7 @@ app_config_parse(struct app_params *app, const char *file_name)
 	APP_PARAM_COUNT(app->pipeline_params, app->n_pipelines);
 
 	/* Save configuration to output file */
-	snprintf(config_file_out,
-		APP_FILE_NAME_SIZE,
-		"%s.out",
-		app->config_file);
-	app_config_save(app, config_file_out);
+	app_config_save(app, app->output_file);
 
 	/* Load TM configuration files */
 	app_config_parse_tm(app);
@@ -2069,6 +2151,20 @@ save_swq_params(struct app_params *app, FILE *f)
 		fprintf(f, "%s = %s\n", "dropless", p->dropless ? "yes" : "no");
 		fprintf(f, "%s = %" PRIu64 "\n", "n_retries", p->n_retries);
 		fprintf(f, "%s = %" PRIu32 "\n", "cpu", p->cpu_socket_id);
+		fprintf(f, "%s = %s\n", "ipv4_frag", p->ipv4_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_frag", p->ipv6_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv4_ras", p->ipv4_ras ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_ras", p->ipv6_ras ? "yes" : "no");
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1)) {
+			fprintf(f, "%s = %" PRIu32 "\n", "mtu", p->mtu);
+			fprintf(f, "%s = %" PRIu32 "\n", "metadata_size", p->metadata_size);
+			fprintf(f, "%s = %s\n",
+				"mempool_direct",
+				app->mempool_params[p->mempool_direct_id].name);
+			fprintf(f, "%s = %s\n",
+				"mempool_indirect",
+				app->mempool_params[p->mempool_indirect_id].name);
+		}
 
 		fputc('\n', f);
 	}
@@ -2360,15 +2456,31 @@ app_config_init(struct app_params *app)
 	return 0;
 }
 
+static char *
+filenamedup(const char *filename, const char *suffix)
+{
+	char *s = malloc(strlen(filename) + strlen(suffix) + 1);
+
+	if (!s)
+		return NULL;
+
+	sprintf(s, "%s%s", filename, suffix);
+	return s;
+}
+
 int
 app_config_args(struct app_params *app, int argc, char **argv)
 {
-	int opt;
-	int option_index, f_present, s_present, p_present, l_present;
+	const char *optname;
+	int opt, option_index;
+	int f_present, s_present, p_present, l_present;
+	int preproc_present, preproc_params_present;
 	int scaned = 0;
 
 	static struct option lgopts[] = {
-		{NULL, 0, 0, 0}
+		{ "preproc", 1, 0, 0 },
+		{ "preproc-args", 1, 0, 0 },
+		{ NULL,  0, 0, 0 }
 	};
 
 	/* Copy application name */
@@ -2378,6 +2490,8 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	s_present = 0;
 	p_present = 0;
 	l_present = 0;
+	preproc_present = 0;
+	preproc_params_present = 0;
 
 	while ((opt = getopt_long(argc, argv, "f:s:p:l:", lgopts,
 			&option_index)) != EOF)
@@ -2443,6 +2557,32 @@ app_config_args(struct app_params *app, int argc, char **argv)
 
 			break;
 
+		case 0:
+			optname = lgopts[option_index].name;
+
+			if (strcmp(optname, "preproc") == 0) {
+				if (preproc_present)
+					rte_panic("Error: Preprocessor argument "
+						"is provided more than once\n");
+				preproc_present = 1;
+
+				app->preproc = strdup(optarg);
+				break;
+			}
+
+			if (strcmp(optname, "preproc-args") == 0) {
+				if (preproc_params_present)
+					rte_panic("Error: Preprocessor args "
+						"are provided more than once\n");
+				preproc_params_present = 1;
+
+				app->preproc_args = strdup(optarg);
+				break;
+			}
+
+			app_print_usage(argv[0]);
+			break;
+
 		default:
 			app_print_usage(argv[0]);
 		}
@@ -2453,5 +2593,40 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	if (!p_present)
 		rte_panic("Error: PORT_MASK is not provided\n");
 
+	/* Check dependencies between args */
+	if (preproc_params_present && (preproc_present == 0))
+		rte_panic("Error: Preprocessor args specified while "
+			"preprocessor is not defined\n");
+
+	app->parser_file = preproc_present ?
+		filenamedup(app->config_file, ".preproc") :
+		strdup(app->config_file);
+	app->output_file = filenamedup(app->config_file, ".out");
+
 	return 0;
 }
+
+int
+app_config_preproc(struct app_params *app)
+{
+	char buffer[256];
+	int status;
+
+	if (app->preproc == NULL)
+		return 0;
+
+	status = access(app->config_file, F_OK | R_OK);
+	APP_CHECK((status == 0), "Unable to open file %s", app->config_file);
+
+	snprintf(buffer, sizeof(buffer), "%s %s %s > %s",
+		app->preproc,
+		app->preproc_args ? app->preproc_args : "",
+		app->config_file,
+		app->parser_file);
+
+	status = system(buffer);
+	APP_CHECK((WIFEXITED(status) && (WEXITSTATUS(status) == 0)),
+		"Error while preprocessing file \"%s\"\n", app->config_file);
+
+	return status;
+}
diff --git a/examples/ip_pipeline/init.c b/examples/ip_pipeline/init.c
index 3f9c68d..46d044e 100644
--- a/examples/ip_pipeline/init.c
+++ b/examples/ip_pipeline/init.c
@@ -803,11 +803,43 @@ app_check_link(struct app_params *app)
 		rte_panic("Some links are DOWN\n");
 }
 
+static uint32_t
+is_any_swq_frag_or_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	for (i = 0; i < app->n_pktq_swq; i++) {
+		struct app_pktq_swq_params *p = &app->swq_params[i];
+
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1) ||
+			(p->ipv4_ras == 1) || (p->ipv6_ras == 1))
+			return 1;
+	}
+
+	return 0;
+}
+
+static void
+app_init_link_frag_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	if (is_any_swq_frag_or_ras(app)) {
+		for (i = 0; i < app->n_pktq_hwq_out; i++) {
+			struct app_pktq_hwq_out_params *p_txq = &app->hwq_out_params[i];
+
+			p_txq->conf.txq_flags &= ~ETH_TXQ_FLAGS_NOMULTSEGS;
+		}
+	}
+}
+
 static void
 app_init_link(struct app_params *app)
 {
 	uint32_t i;
 
+	app_init_link_frag_ras(app);
+
 	for (i = 0; i < app->n_links; i++) {
 		struct app_link_params *p_link = &app->link_params[i];
 		uint32_t link_id, n_hwq_in, n_hwq_out, j;
@@ -916,13 +948,19 @@ app_init_swq(struct app_params *app)
 
 	for (i = 0; i < app->n_pktq_swq; i++) {
 		struct app_pktq_swq_params *p = &app->swq_params[i];
+		unsigned flags = 0;
+
+		if (app_swq_get_readers(app, p) == 1)
+			flags |= RING_F_SC_DEQ;
+		if (app_swq_get_writers(app, p) == 1)
+			flags |= RING_F_SP_ENQ;
 
 		APP_LOG(app, HIGH, "Initializing %s...", p->name);
 		app->swq[i] = rte_ring_create(
 				p->name,
 				p->size,
 				p->cpu_socket_id,
-				RING_F_SP_ENQ | RING_F_SC_DEQ);
+				flags);
 
 		if (app->swq[i] == NULL)
 			rte_panic("%s init error\n", p->name);
@@ -1059,11 +1097,50 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_IN_SWQ:
-			out->type = PIPELINE_PORT_IN_RING_READER;
-			out->params.ring.ring = app->swq[in->id];
-			out->burst_size = app->swq_params[in->id].burst_read;
-			/* What about frag and ras ports? */
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_frag == 0) && (swq_params->ipv6_frag == 0)) {
+				if (app_swq_get_readers(app, swq_params) == 1) {
+					out->type = PIPELINE_PORT_IN_RING_READER;
+					out->params.ring.ring = app->swq[in->id];
+					out->burst_size = app->swq_params[in->id].burst_read;
+				} else {
+					out->type = PIPELINE_PORT_IN_RING_MULTI_READER;
+					out->params.ring_multi.ring = app->swq[in->id];
+					out->burst_size = swq_params->burst_read;
+				}
+			} else {
+				if (swq_params->ipv4_frag == 1) {
+					struct rte_port_ring_reader_ipv4_frag_params *params =
+						&out->params.ring_ipv4_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV4_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				} else {
+					struct rte_port_ring_reader_ipv6_frag_params *params =
+						&out->params.ring_ipv6_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV6_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				}
+			}
 			break;
+		}
 		case APP_PKTQ_IN_TM:
 			out->type = PIPELINE_PORT_IN_SCHED_READER;
 			out->params.sched.sched = app->tm[in->id];
@@ -1122,28 +1199,68 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_OUT_SWQ:
-			if (app->swq_params[in->id].dropless == 0) {
-				struct rte_port_ring_writer_params *params =
-					&out->params.ring;
-
-				out->type = PIPELINE_PORT_OUT_RING_WRITER;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_ras == 0) && (swq_params->ipv6_ras == 0)) {
+				if (app_swq_get_writers(app, swq_params) == 1) {
+					if (app->swq_params[in->id].dropless == 0) {
+						struct rte_port_ring_writer_params *params =
+							&out->params.ring;
+
+						out->type = PIPELINE_PORT_OUT_RING_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+					} else {
+						struct rte_port_ring_writer_nodrop_params
+							*params = &out->params.ring_nodrop;
+
+						out->type =
+							PIPELINE_PORT_OUT_RING_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+						params->n_retries =
+							app->swq_params[in->id].n_retries;
+					}
+				} else {
+					if (swq_params->dropless == 0) {
+						struct rte_port_ring_multi_writer_params *params =
+							&out->params.ring_multi;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+					} else {
+						struct rte_port_ring_multi_writer_nodrop_params
+							*params = &out->params.ring_multi_nodrop;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+						params->n_retries = swq_params->n_retries;
+					}
+				}
 			} else {
-				struct rte_port_ring_writer_nodrop_params
-					*params = &out->params.ring_nodrop;
-
-				out->type =
-					PIPELINE_PORT_OUT_RING_WRITER_NODROP;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
-				params->n_retries =
-					app->swq_params[in->id].n_retries;
+				if (swq_params->ipv4_ras == 1) {
+					struct rte_port_ring_writer_ipv4_ras_params *params =
+						&out->params.ring_ipv4_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				} else {
+					struct rte_port_ring_writer_ipv6_ras_params *params =
+						&out->params.ring_ipv6_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				}
 			}
-			/* What about frag and ras ports? */
 			break;
+		}
 		case APP_PKTQ_OUT_TM: {
 			struct rte_port_sched_writer_params *params =
 				&out->params.sched;
diff --git a/examples/ip_pipeline/main.c b/examples/ip_pipeline/main.c
index 862e2f2..4944dcf 100644
--- a/examples/ip_pipeline/main.c
+++ b/examples/ip_pipeline/main.c
@@ -45,7 +45,9 @@ main(int argc, char **argv)
 
 	app_config_args(&app, argc, argv);
 
-	app_config_parse(&app, app.config_file);
+	app_config_preproc(&app);
+
+	app_config_parse(&app, app.parser_file);
 
 	app_config_check(&app);
 
diff --git a/examples/ip_pipeline/pipeline_be.h b/examples/ip_pipeline/pipeline_be.h
index 51f1e4f..f7269c0 100644
--- a/examples/ip_pipeline/pipeline_be.h
+++ b/examples/ip_pipeline/pipeline_be.h
@@ -45,6 +45,7 @@
 enum pipeline_port_in_type {
 	PIPELINE_PORT_IN_ETHDEV_READER,
 	PIPELINE_PORT_IN_RING_READER,
+	PIPELINE_PORT_IN_RING_MULTI_READER,
 	PIPELINE_PORT_IN_RING_READER_IPV4_FRAG,
 	PIPELINE_PORT_IN_RING_READER_IPV6_FRAG,
 	PIPELINE_PORT_IN_SCHED_READER,
@@ -56,6 +57,7 @@ struct pipeline_port_in_params {
 	union {
 		struct rte_port_ethdev_reader_params ethdev;
 		struct rte_port_ring_reader_params ring;
+		struct rte_port_ring_multi_reader_params ring_multi;
 		struct rte_port_ring_reader_ipv4_frag_params ring_ipv4_frag;
 		struct rte_port_ring_reader_ipv6_frag_params ring_ipv6_frag;
 		struct rte_port_sched_reader_params sched;
@@ -72,6 +74,8 @@ pipeline_port_in_params_convert(struct pipeline_port_in_params  *p)
 		return (void *) &p->params.ethdev;
 	case PIPELINE_PORT_IN_RING_READER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return (void *) &p->params.ring_ipv4_frag;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -93,6 +97,8 @@ pipeline_port_in_params_get_ops(struct pipeline_port_in_params  *p)
 		return &rte_port_ethdev_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER:
 		return &rte_port_ring_reader_ops;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return &rte_port_ring_multi_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return &rte_port_ring_reader_ipv4_frag_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -110,7 +116,9 @@ enum pipeline_port_out_type {
 	PIPELINE_PORT_OUT_ETHDEV_WRITER,
 	PIPELINE_PORT_OUT_ETHDEV_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER,
 	PIPELINE_PORT_OUT_RING_WRITER_NODROP,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS,
 	PIPELINE_PORT_OUT_SCHED_WRITER,
@@ -123,7 +131,9 @@ struct pipeline_port_out_params {
 		struct rte_port_ethdev_writer_params ethdev;
 		struct rte_port_ethdev_writer_nodrop_params ethdev_nodrop;
 		struct rte_port_ring_writer_params ring;
+		struct rte_port_ring_multi_writer_params ring_multi;
 		struct rte_port_ring_writer_nodrop_params ring_nodrop;
+		struct rte_port_ring_multi_writer_nodrop_params ring_multi_nodrop;
 		struct rte_port_ring_writer_ipv4_ras_params ring_ipv4_ras;
 		struct rte_port_ring_writer_ipv6_ras_params ring_ipv6_ras;
 		struct rte_port_sched_writer_params sched;
@@ -140,8 +150,12 @@ pipeline_port_out_params_convert(struct pipeline_port_out_params  *p)
 		return (void *) &p->params.ethdev_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return (void *) &p->params.ring_nodrop;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return (void *) &p->params.ring_multi_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return (void *) &p->params.ring_ipv4_ras;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
@@ -164,8 +178,12 @@ pipeline_port_out_params_get_ops(struct pipeline_port_out_params  *p)
 		return &rte_port_ethdev_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return &rte_port_ring_writer_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return &rte_port_ring_multi_writer_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return &rte_port_ring_writer_nodrop_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return &rte_port_ring_multi_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return &rte_port_ring_writer_ipv4_ras_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
-- 
1.7.9.5

Re: [dpdk-dev] [PATCH v2 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[Dumitrescu, Cristian]

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Piotr Azarewicz
> Sent: Thursday, September 24, 2015 12:56 PM
> To: dev@dpdk.org
> Subject: [dpdk-dev] [PATCH v2 0/3] ip_pipeline: add MP/MC and frag/ras
> support to SWQs
> 
> 
> v2 changes:
> - rte_port_ring:
> 	- fixed checkpatch errors
> 	- interlace the implementation of multi into the implementation of
> single
> 	- reduced the amount of code duplication
> 

Acked-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>

Re: [dpdk-dev] [PATCH v2 2/3] port: fix ras ring ports

[Thomas Monjalon]

2015-09-24 11:55, Piotr Azarewicz:
> Bug fixes for ring ports with IPv4/IPv6 reassembly support.
> Previous implementation can't work properly due to incorrect choosing
> process function.
> Also, assuming that, when processing ip packet, ip header is know we can
> set l3_len parameter here.
> 
> Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>

Please add a "Fixes:" tag.
Thanks

Re: [dpdk-dev] [PATCH v2 1/3] port: add mp/mc ring ports

[Thomas Monjalon]

2015-09-24 11:55, Piotr Azarewicz:
> ring_multi_reader input port (on top of multi consumer rte_ring)
> ring_multi_writer output port (on top of multi producer rte_ring)
> 
> Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
> ---
>  lib/librte_port/rte_port_ring.c |  311 +++++++++++++++++++++++++++++++++++----
>  lib/librte_port/rte_port_ring.h |   35 ++++-

Piotr,
You forgot the shared library case: the new functions are missing in the .map.

Cristian,
As a maintainer, please check compilation of shared library.

See you for the v3 ;)
Thanks

[dpdk-dev] [PATCH v3 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[Piotr Azarewicz]

This patch set enhancement ip_pipeline application:
- librte_port: add support for multi-producer/multi-consumer ring ports
- librte_port: bug fixes for ring ports with IPv4/IPv6 reassembly support
- ip_pipeline application: integrate MP/MC and fragmentation/reassembly support to SWQs

v2 changes:
- rte_port_ring:
	- fixed checkpatch errors
	- interlace the implementation of multi into the implementation of single
	- reduced the amount of code duplication

v3 changes:
- new functions added in the .map
- add a "Fixes:" tag in commit comment

Acked-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>

Piotr Azarewicz (3):
  port: add mp/mc ring ports
  port: fix ras ring ports
  examples/ip_pipeline: add mp/mc and frag/ras swq

 examples/ip_pipeline/app.h           |   14 ++
 examples/ip_pipeline/config_check.c  |   45 ++++-
 examples/ip_pipeline/config_parse.c  |  195 +++++++++++++++++++--
 examples/ip_pipeline/init.c          |  165 +++++++++++++++---
 examples/ip_pipeline/main.c          |    4 +-
 examples/ip_pipeline/pipeline_be.h   |   18 ++
 lib/librte_port/rte_port_ras.c       |    8 +-
 lib/librte_port/rte_port_ring.c      |  311 +++++++++++++++++++++++++++++++---
 lib/librte_port/rte_port_ring.h      |   35 +++-
 lib/librte_port/rte_port_version.map |    9 +
 10 files changed, 733 insertions(+), 71 deletions(-)

-- 
1.7.9.5

[dpdk-dev] [PATCH v3 1/3] port: add mp/mc ring ports

[Piotr Azarewicz]

ring_multi_reader input port (on top of multi consumer rte_ring)
ring_multi_writer output port (on top of multi producer rte_ring)

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_ring.c      |  311 +++++++++++++++++++++++++++++++---
 lib/librte_port/rte_port_ring.h      |   35 +++-
 lib/librte_port/rte_port_version.map |    9 +
 3 files changed, 325 insertions(+), 30 deletions(-)

diff --git a/lib/librte_port/rte_port_ring.c b/lib/librte_port/rte_port_ring.c
index 9461c05..755dfc1 100644
--- a/lib/librte_port/rte_port_ring.c
+++ b/lib/librte_port/rte_port_ring.c
@@ -63,15 +63,19 @@ struct rte_port_ring_reader {
 };
 
 static void *
-rte_port_ring_reader_create(void *params, int socket_id)
+rte_port_ring_reader_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_reader_params *conf =
 			(struct rte_port_ring_reader_params *) params;
 	struct rte_port_ring_reader *port;
 
 	/* Check input parameters */
-	if (conf == NULL) {
-		RTE_LOG(ERR, PORT, "%s: params is NULL\n", __func__);
+	if ((conf == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->cons.sc_dequeue && is_multi) ||
+		(!(conf->ring->cons.sc_dequeue) && !is_multi)) {
+		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
 	}
 
@@ -89,6 +93,18 @@ rte_port_ring_reader_create(void *params, int socket_id)
 	return port;
 }
 
+static void *
+rte_port_ring_reader_create(void *params, int socket_id)
+{
+	return rte_port_ring_reader_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_reader_create(void *params, int socket_id)
+{
+	return rte_port_ring_reader_create_internal(params, socket_id, 1);
+}
+
 static int
 rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
 {
@@ -102,6 +118,19 @@ rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
 }
 
 static int
+rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts,
+	uint32_t n_pkts)
+{
+	struct rte_port_ring_reader *p = (struct rte_port_ring_reader *) port;
+	uint32_t nb_rx;
+
+	nb_rx = rte_ring_mc_dequeue_burst(p->ring, (void **) pkts, n_pkts);
+	RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx);
+
+	return nb_rx;
+}
+
+static int
 rte_port_ring_reader_free(void *port)
 {
 	if (port == NULL) {
@@ -155,10 +184,12 @@ struct rte_port_ring_writer {
 	uint32_t tx_burst_sz;
 	uint32_t tx_buf_count;
 	uint64_t bsz_mask;
+	uint32_t is_multi;
 };
 
 static void *
-rte_port_ring_writer_create(void *params, int socket_id)
+rte_port_ring_writer_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_writer_params *conf =
 			(struct rte_port_ring_writer_params *) params;
@@ -166,7 +197,9 @@ rte_port_ring_writer_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue && is_multi) ||
+		(!(conf->ring->prod.sp_enqueue) && !is_multi) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -185,10 +218,23 @@ rte_port_ring_writer_create(void *params, int socket_id)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+	port->is_multi = is_multi;
 
 	return port;
 }
 
+static void *
+rte_port_ring_writer_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_writer_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_create_internal(params, socket_id, 1);
+}
+
 static inline void
 send_burst(struct rte_port_ring_writer *p)
 {
@@ -204,6 +250,21 @@ send_burst(struct rte_port_ring_writer *p)
 	p->tx_buf_count = 0;
 }
 
+static inline void
+send_burst_mp(struct rte_port_ring_writer *p)
+{
+	uint32_t nb_tx;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+			p->tx_buf_count);
+
+	RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
 static int
 rte_port_ring_writer_tx(void *port, struct rte_mbuf *pkt)
 {
@@ -218,9 +279,23 @@ rte_port_ring_writer_tx(void *port, struct rte_mbuf *pkt)
 }
 
 static int
-rte_port_ring_writer_tx_bulk(void *port,
+rte_port_ring_multi_writer_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static inline int __attribute__((always_inline))
+rte_port_ring_writer_tx_bulk_internal(void *port,
 		struct rte_mbuf **pkts,
-		uint64_t pkts_mask)
+		uint64_t pkts_mask,
+		uint32_t is_multi)
 {
 	struct rte_port_ring_writer *p =
 		(struct rte_port_ring_writer *) port;
@@ -234,11 +309,20 @@ rte_port_ring_writer_tx_bulk(void *port,
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t n_pkts_ok;
 
-		if (tx_buf_count)
-			send_burst(p);
+		if (tx_buf_count) {
+			if (is_multi)
+				send_burst_mp(p);
+			else
+				send_burst(p);
+		}
 
 		RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, n_pkts);
-		n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		if (is_multi)
+			n_pkts_ok = rte_ring_mp_enqueue_burst(p->ring, (void **)pkts,
+				n_pkts);
+		else
+			n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts,
+				n_pkts);
 
 		RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, n_pkts - n_pkts_ok);
 		for ( ; n_pkts_ok < n_pkts; n_pkts_ok++) {
@@ -258,14 +342,34 @@ rte_port_ring_writer_tx_bulk(void *port,
 		}
 
 		p->tx_buf_count = tx_buf_count;
-		if (tx_buf_count >= p->tx_burst_sz)
-			send_burst(p);
+		if (tx_buf_count >= p->tx_burst_sz) {
+			if (is_multi)
+				send_burst_mp(p);
+			else
+				send_burst(p);
+		}
 	}
 
 	return 0;
 }
 
 static int
+rte_port_ring_writer_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 0);
+}
+
+static int
+rte_port_ring_multi_writer_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 1);
+}
+
+static int
 rte_port_ring_writer_flush(void *port)
 {
 	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
@@ -277,14 +381,31 @@ rte_port_ring_writer_flush(void *port)
 }
 
 static int
+rte_port_ring_multi_writer_flush(void *port)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static int
 rte_port_ring_writer_free(void *port)
 {
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
 	if (port == NULL) {
 		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
 		return -EINVAL;
 	}
 
-	rte_port_ring_writer_flush(port);
+	if (p->is_multi)
+		rte_port_ring_multi_writer_flush(port);
+	else
+		rte_port_ring_writer_flush(port);
+
 	rte_free(port);
 
 	return 0;
@@ -332,10 +453,12 @@ struct rte_port_ring_writer_nodrop {
 	uint32_t tx_buf_count;
 	uint64_t bsz_mask;
 	uint64_t n_retries;
+	uint32_t is_multi;
 };
 
 static void *
-rte_port_ring_writer_nodrop_create(void *params, int socket_id)
+rte_port_ring_writer_nodrop_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_writer_nodrop_params *conf =
 			(struct rte_port_ring_writer_nodrop_params *) params;
@@ -343,7 +466,9 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue && is_multi) ||
+		(!(conf->ring->prod.sp_enqueue) && !is_multi) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -362,6 +487,7 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+	port->is_multi = is_multi;
 
 	/*
 	 * When n_retries is 0 it means that we should wait for every packet to
@@ -373,6 +499,18 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 	return port;
 }
 
+static void *
+rte_port_ring_writer_nodrop_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_writer_nodrop_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 1);
+}
+
 static inline void
 send_burst_nodrop(struct rte_port_ring_writer_nodrop *p)
 {
@@ -402,6 +540,35 @@ send_burst_nodrop(struct rte_port_ring_writer_nodrop *p)
 	p->tx_buf_count = 0;
 }
 
+static inline void
+send_burst_mp_nodrop(struct rte_port_ring_writer_nodrop *p)
+{
+	uint32_t nb_tx = 0, i;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+				p->tx_buf_count);
+
+	/* We sent all the packets in a first try */
+	if (nb_tx >= p->tx_buf_count)
+		return;
+
+	for (i = 0; i < p->n_retries; i++) {
+		nb_tx += rte_ring_mp_enqueue_burst(p->ring,
+				(void **) (p->tx_buf + nb_tx), p->tx_buf_count - nb_tx);
+
+		/* We sent all the packets in more than one try */
+		if (nb_tx >= p->tx_buf_count)
+			return;
+	}
+
+	/* We didn't send the packets in maximum allowed attempts */
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
 static int
 rte_port_ring_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
 {
@@ -417,9 +584,24 @@ rte_port_ring_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
 }
 
 static int
-rte_port_ring_writer_nodrop_tx_bulk(void *port,
+rte_port_ring_multi_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static inline int __attribute__((always_inline))
+rte_port_ring_writer_nodrop_tx_bulk_internal(void *port,
 		struct rte_mbuf **pkts,
-		uint64_t pkts_mask)
+		uint64_t pkts_mask,
+		uint32_t is_multi)
 {
 	struct rte_port_ring_writer_nodrop *p =
 		(struct rte_port_ring_writer_nodrop *) port;
@@ -433,24 +615,37 @@ rte_port_ring_writer_nodrop_tx_bulk(void *port,
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t n_pkts_ok;
 
-		if (tx_buf_count)
-			send_burst_nodrop(p);
+		if (tx_buf_count) {
+			if (is_multi)
+				send_burst_mp_nodrop(p);
+			else
+				send_burst_nodrop(p);
+		}
 
 		RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, n_pkts);
-		n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		if (is_multi)
+			n_pkts_ok =
+				rte_ring_mp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		else
+			n_pkts_ok =
+				rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
 
 		if (n_pkts_ok >= n_pkts)
 			return 0;
 
 		/*
-		 * If we didnt manage to send all packets in single burst, move
+		 * If we didn't manage to send all packets in single burst, move
 		 * remaining packets to the buffer and call send burst.
 		 */
 		for (; n_pkts_ok < n_pkts; n_pkts_ok++) {
 			struct rte_mbuf *pkt = pkts[n_pkts_ok];
+
 			p->tx_buf[p->tx_buf_count++] = pkt;
 		}
-		send_burst_nodrop(p);
+		if (is_multi)
+			send_burst_mp_nodrop(p);
+		else
+			send_burst_nodrop(p);
 	} else {
 		for ( ; pkts_mask; ) {
 			uint32_t pkt_index = __builtin_ctzll(pkts_mask);
@@ -463,14 +658,36 @@ rte_port_ring_writer_nodrop_tx_bulk(void *port,
 		}
 
 		p->tx_buf_count = tx_buf_count;
-		if (tx_buf_count >= p->tx_burst_sz)
-			send_burst_nodrop(p);
+		if (tx_buf_count >= p->tx_burst_sz) {
+			if (is_multi)
+				send_burst_mp_nodrop(p);
+			else
+				send_burst_nodrop(p);
+		}
 	}
 
 	return 0;
 }
 
 static int
+rte_port_ring_writer_nodrop_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return
+		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 0);
+}
+
+static int
+rte_port_ring_multi_writer_nodrop_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return
+		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 1);
+}
+
+static int
 rte_port_ring_writer_nodrop_flush(void *port)
 {
 	struct rte_port_ring_writer_nodrop *p =
@@ -483,14 +700,33 @@ rte_port_ring_writer_nodrop_flush(void *port)
 }
 
 static int
+rte_port_ring_multi_writer_nodrop_flush(void *port)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static int
 rte_port_ring_writer_nodrop_free(void *port)
 {
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
 	if (port == NULL) {
 		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
 		return -EINVAL;
 	}
 
-	rte_port_ring_writer_nodrop_flush(port);
+	if (p->is_multi)
+		rte_port_ring_multi_writer_nodrop_flush(port);
+	else
+		rte_port_ring_writer_nodrop_flush(port);
+
 	rte_free(port);
 
 	return 0;
@@ -539,3 +775,28 @@ struct rte_port_out_ops rte_port_ring_writer_nodrop_ops = {
 	.f_flush = rte_port_ring_writer_nodrop_flush,
 	.f_stats = rte_port_ring_writer_nodrop_stats_read,
 };
+
+struct rte_port_in_ops rte_port_ring_multi_reader_ops = {
+	.f_create = rte_port_ring_multi_reader_create,
+	.f_free = rte_port_ring_reader_free,
+	.f_rx = rte_port_ring_multi_reader_rx,
+	.f_stats = rte_port_ring_reader_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_ops = {
+	.f_create = rte_port_ring_multi_writer_create,
+	.f_free = rte_port_ring_writer_free,
+	.f_tx = rte_port_ring_multi_writer_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_flush,
+	.f_stats = rte_port_ring_writer_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops = {
+	.f_create = rte_port_ring_multi_writer_nodrop_create,
+	.f_free = rte_port_ring_writer_nodrop_free,
+	.f_tx = rte_port_ring_multi_writer_nodrop_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_nodrop_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_nodrop_flush,
+	.f_stats = rte_port_ring_writer_nodrop_stats_read,
+};
diff --git a/lib/librte_port/rte_port_ring.h b/lib/librte_port/rte_port_ring.h
index 89a219b..de377d2 100644
--- a/lib/librte_port/rte_port_ring.h
+++ b/lib/librte_port/rte_port_ring.h
@@ -42,8 +42,14 @@ extern "C" {
  * @file
  * RTE Port Ring
  *
- * ring_reader: input port built on top of pre-initialized single consumer ring
- * ring_writer: output port built on top of pre-initialized single producer ring
+ * ring_reader:
+ *      input port built on top of pre-initialized single consumer ring
+ * ring_writer:
+ *      output port built on top of pre-initialized single producer ring
+ * ring_multi_reader:
+ *      input port built on top of pre-initialized multi consumers ring
+ * ring_multi_writer:
+ *      output port built on top of pre-initialized multi producers ring
  *
  ***/
 
@@ -55,7 +61,7 @@ extern "C" {
 
 /** ring_reader port parameters */
 struct rte_port_ring_reader_params {
-	/** Underlying single consumer ring that has to be pre-initialized */
+	/** Underlying consumer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 };
 
@@ -64,7 +70,7 @@ extern struct rte_port_in_ops rte_port_ring_reader_ops;
 
 /** ring_writer port parameters */
 struct rte_port_ring_writer_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -77,7 +83,7 @@ extern struct rte_port_out_ops rte_port_ring_writer_ops;
 
 /** ring_writer_nodrop port parameters */
 struct rte_port_ring_writer_nodrop_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -91,6 +97,25 @@ struct rte_port_ring_writer_nodrop_params {
 /** ring_writer_nodrop port operations */
 extern struct rte_port_out_ops rte_port_ring_writer_nodrop_ops;
 
+/** ring_multi_reader port parameters */
+#define rte_port_ring_multi_reader_params rte_port_ring_reader_params
+
+/** ring_multi_reader port operations */
+extern struct rte_port_in_ops rte_port_ring_multi_reader_ops;
+
+/** ring_multi_writer port parameters */
+#define rte_port_ring_multi_writer_params rte_port_ring_writer_params
+
+/** ring_multi_writer port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_ops;
+
+/** ring_multi_writer_nodrop port parameters */
+#define rte_port_ring_multi_writer_nodrop_params \
+	rte_port_ring_writer_nodrop_params
+
+/** ring_multi_writer_nodrop port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops;
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/librte_port/rte_port_version.map b/lib/librte_port/rte_port_version.map
index 4e49ff6..7a0b34d 100644
--- a/lib/librte_port/rte_port_version.map
+++ b/lib/librte_port/rte_port_version.map
@@ -26,3 +26,12 @@ DPDK_2.1 {
 	rte_port_ring_writer_nodrop_ops;
 
 } DPDK_2.0;
+
+DPDK_2.2 {
+	global:
+
+	rte_port_ring_multi_reader_ops;
+	rte_port_ring_multi_writer_ops;
+	rte_port_ring_multi_writer_nodrop_ops;
+
+} DPDK_2.1;
-- 
1.7.9.5

[dpdk-dev] [PATCH v3 2/3] port: fix ras ring ports

[Piotr Azarewicz]

Bug fixes for ring ports with IPv4/IPv6 reassembly support.
Previous implementation can't work properly due to incorrect choosing
process function.
Also, assuming that, when processing ip packet, ip header is know we can
set l3_len parameter here.

Fixes: 50f54a84dfb7 ("port: add IPv6 reassembly port")

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_ras.c |    8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/lib/librte_port/rte_port_ras.c b/lib/librte_port/rte_port_ras.c
index 6bd0f8c..e45d450 100644
--- a/lib/librte_port/rte_port_ras.c
+++ b/lib/librte_port/rte_port_ras.c
@@ -144,7 +144,7 @@ rte_port_ring_writer_ras_create(void *params, int socket_id, int is_ipv4)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 
-	port->f_ras = (is_ipv4 == 0) ? process_ipv4 : process_ipv6;
+	port->f_ras = (is_ipv4 == 1) ? process_ipv4 : process_ipv6;
 
 	return port;
 }
@@ -182,7 +182,7 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 	/* Assume there is no ethernet header */
 	struct ipv4_hdr *pkt_hdr = rte_pktmbuf_mtod(pkt, struct ipv4_hdr *);
 
-	/* Get "Do not fragment" flag and fragment offset */
+	/* Get "More fragments" flag and fragment offset */
 	uint16_t frag_field = rte_be_to_cpu_16(pkt_hdr->fragment_offset);
 	uint16_t frag_offset = (uint16_t)(frag_field & IPV4_HDR_OFFSET_MASK);
 	uint16_t frag_flag = (uint16_t)(frag_field & IPV4_HDR_MF_FLAG);
@@ -195,6 +195,8 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv4_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(),
 				pkt_hdr);
@@ -224,6 +226,8 @@ process_ipv6(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr) + sizeof(*frag_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv6_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(), pkt_hdr,
 				frag_hdr);
-- 
1.7.9.5

[dpdk-dev] [PATCH v3 3/3] examples/ip_pipeline: add mp/mc and frag/ras swq

[Piotr Azarewicz]

Add integrated MP/MC and fragmentation/reassembly support to SWQs

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 examples/ip_pipeline/app.h          |   14 +++
 examples/ip_pipeline/config_check.c |   45 +++++++-
 examples/ip_pipeline/config_parse.c |  195 +++++++++++++++++++++++++++++++++--
 examples/ip_pipeline/init.c         |  165 ++++++++++++++++++++++++-----
 examples/ip_pipeline/main.c         |    4 +-
 examples/ip_pipeline/pipeline_be.h  |   18 ++++
 6 files changed, 402 insertions(+), 39 deletions(-)

diff --git a/examples/ip_pipeline/app.h b/examples/ip_pipeline/app.h
index 521e3a0..943466e 100644
--- a/examples/ip_pipeline/app.h
+++ b/examples/ip_pipeline/app.h
@@ -107,6 +107,14 @@ struct app_pktq_swq_params {
 	uint32_t dropless;
 	uint64_t n_retries;
 	uint32_t cpu_socket_id;
+	uint32_t ipv4_frag;
+	uint32_t ipv6_frag;
+	uint32_t ipv4_ras;
+	uint32_t ipv6_ras;
+	uint32_t mtu;
+	uint32_t metadata_size;
+	uint32_t mempool_direct_id;
+	uint32_t mempool_indirect_id;
 };
 
 #ifndef APP_FILE_NAME_SIZE
@@ -405,6 +413,10 @@ struct app_params {
 	char app_name[APP_APPNAME_SIZE];
 	const char *config_file;
 	const char *script_file;
+	const char *parser_file;
+	const char *output_file;
+	const char *preproc;
+	const char *preproc_args;
 	uint64_t port_mask;
 	uint32_t log_level;
 
@@ -880,6 +892,8 @@ int app_config_init(struct app_params *app);
 int app_config_args(struct app_params *app,
 	int argc, char **argv);
 
+int app_config_preproc(struct app_params *app);
+
 int app_config_parse(struct app_params *app,
 	const char *file_name);
 
diff --git a/examples/ip_pipeline/config_check.c b/examples/ip_pipeline/config_check.c
index 07f4c8b..8052bc4 100644
--- a/examples/ip_pipeline/config_check.c
+++ b/examples/ip_pipeline/config_check.c
@@ -33,6 +33,8 @@
 
 #include <stdio.h>
 
+#include <rte_ip.h>
+
 #include "app.h"
 
 static void
@@ -193,6 +195,7 @@ check_swqs(struct app_params *app)
 		struct app_pktq_swq_params *p = &app->swq_params[i];
 		uint32_t n_readers = app_swq_get_readers(app, p);
 		uint32_t n_writers = app_swq_get_writers(app, p);
+		uint32_t n_flags;
 
 		APP_CHECK((p->size > 0),
 			"%s size is 0\n", p->name);
@@ -217,14 +220,48 @@ check_swqs(struct app_params *app)
 		APP_CHECK((n_readers != 0),
 			"%s has no reader\n", p->name);
 
-		APP_CHECK((n_readers == 1),
-			"%s has more than one reader\n", p->name);
+		if (n_readers > 1)
+			APP_LOG(app, LOW, "%s has more than one reader", p->name);
 
 		APP_CHECK((n_writers != 0),
 			"%s has no writer\n", p->name);
 
-		APP_CHECK((n_writers == 1),
-			"%s has more than one writer\n", p->name);
+		if (n_writers > 1)
+			APP_LOG(app, LOW, "%s has more than one writer", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag + p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((n_flags < 2),
+			"%s has more than one fragmentation or reassembly mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_readers > 1) && (n_flags == 1))),
+			"%s has more than one reader when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_writers > 1) && (n_flags == 1))),
+			"%s has more than one writer when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		n_flags = p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((!((p->dropless == 1) && (n_flags == 1))),
+			"%s has dropless when reassembly mode enabled\n", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag;
+
+		if (n_flags == 1) {
+			uint16_t ip_hdr_size = (p->ipv4_frag) ? sizeof(struct ipv4_hdr) :
+				sizeof(struct ipv6_hdr);
+
+			APP_CHECK((p->mtu > ip_hdr_size),
+				"%s mtu size is smaller than ip header\n", p->name);
+
+			APP_CHECK((!((p->mtu - ip_hdr_size) % 8)),
+				"%s mtu size is incorrect\n", p->name);
+		}
 	}
 }
 
diff --git a/examples/ip_pipeline/config_parse.c b/examples/ip_pipeline/config_parse.c
index c9b78f9..a35bd3e 100644
--- a/examples/ip_pipeline/config_parse.c
+++ b/examples/ip_pipeline/config_parse.c
@@ -156,6 +156,14 @@ static const struct app_pktq_swq_params default_swq_params = {
 	.dropless = 0,
 	.n_retries = 0,
 	.cpu_socket_id = 0,
+	.ipv4_frag = 0,
+	.ipv6_frag = 0,
+	.ipv4_ras = 0,
+	.ipv6_ras = 0,
+	.mtu = 0,
+	.metadata_size = 0,
+	.mempool_direct_id = 0,
+	.mempool_indirect_id = 0,
 };
 
 struct app_pktq_tm_params default_tm_params = {
@@ -196,13 +204,15 @@ struct app_pipeline_params default_pipeline_params = {
 
 static const char app_usage[] =
 	"Usage: %s [-f CONFIG_FILE] [-s SCRIPT_FILE] -p PORT_MASK "
-	"[-l LOG_LEVEL]\n"
+	"[-l LOG_LEVEL] [--preproc PREPROCESSOR] [--preproc-args ARGS]\n"
 	"\n"
 	"Arguments:\n"
 	"\t-f CONFIG_FILE: Default config file is %s\n"
 	"\t-p PORT_MASK: Mask of NIC port IDs in hexadecimal format\n"
 	"\t-s SCRIPT_FILE: No CLI script file is run when not specified\n"
 	"\t-l LOG_LEVEL: 0 = NONE, 1 = HIGH PRIO (default), 2 = LOW PRIO\n"
+	"\t--preproc PREPROCESSOR: Configuration file pre-processor\n"
+	"\t--preproc-args ARGS: Arguments to be passed to pre-processor\n"
 	"\n";
 
 static void
@@ -1107,6 +1117,10 @@ parse_pipeline(struct app_params *app,
 			ret = parser_read_uint32(&param->timer_period,
 				ent->value);
 		else {
+			APP_CHECK((param->n_args < APP_MAX_PIPELINE_ARGS),
+				"CFG: [%s] out of memory",
+				section_name);
+
 			param->args_name[param->n_args] = strdup(ent->name);
 			param->args_value[param->n_args] = strdup(ent->value);
 
@@ -1397,6 +1411,7 @@ parse_swq(struct app_params *app,
 	struct app_pktq_swq_params *param;
 	struct rte_cfgfile_entry *entries;
 	int n_entries, ret, i;
+	unsigned frag_entries = 0;
 	ssize_t param_idx;
 
 	n_entries = rte_cfgfile_section_num_entries(cfg, section_name);
@@ -1438,6 +1453,71 @@ parse_swq(struct app_params *app,
 		else if (strcmp(ent->name, "cpu") == 0)
 			ret = parser_read_uint32(&param->cpu_socket_id,
 				ent->value);
+		else if (strcmp(ent->name, "ipv4_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1500;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1320;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv4_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "mtu") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->mtu,
+				ent->value);
+		} else if (strcmp(ent->name, "metadata_size") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->metadata_size,
+				ent->value);
+		} else if (strcmp(ent->name, "mempool_direct") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_direct_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		} else if (strcmp(ent->name, "mempool_indirect") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_indirect_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		}
 
 		APP_CHECK(ret != -ESRCH,
 			"CFG: [%s] entry '%s': unknown entry\n",
@@ -1450,6 +1530,13 @@ parse_swq(struct app_params *app,
 			ent->value);
 	}
 
+	if (frag_entries == 1) {
+		APP_CHECK(((param->ipv4_frag == 1) || (param->ipv6_frag == 1)),
+			"CFG: [%s] ipv4/ipv6 frag is off : unsupported entries on this"
+			" configuration\n",
+			section_name);
+	}
+
 	free(entries);
 }
 
@@ -1769,7 +1856,6 @@ parse_port_mask(struct app_params *app, uint64_t port_mask)
 int
 app_config_parse(struct app_params *app, const char *file_name)
 {
-	char config_file_out[APP_FILE_NAME_SIZE];
 	struct rte_cfgfile *cfg;
 	char **section_names;
 	int i, j, sect_count;
@@ -1851,11 +1937,7 @@ app_config_parse(struct app_params *app, const char *file_name)
 	APP_PARAM_COUNT(app->pipeline_params, app->n_pipelines);
 
 	/* Save configuration to output file */
-	snprintf(config_file_out,
-		APP_FILE_NAME_SIZE,
-		"%s.out",
-		app->config_file);
-	app_config_save(app, config_file_out);
+	app_config_save(app, app->output_file);
 
 	/* Load TM configuration files */
 	app_config_parse_tm(app);
@@ -2069,6 +2151,20 @@ save_swq_params(struct app_params *app, FILE *f)
 		fprintf(f, "%s = %s\n", "dropless", p->dropless ? "yes" : "no");
 		fprintf(f, "%s = %" PRIu64 "\n", "n_retries", p->n_retries);
 		fprintf(f, "%s = %" PRIu32 "\n", "cpu", p->cpu_socket_id);
+		fprintf(f, "%s = %s\n", "ipv4_frag", p->ipv4_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_frag", p->ipv6_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv4_ras", p->ipv4_ras ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_ras", p->ipv6_ras ? "yes" : "no");
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1)) {
+			fprintf(f, "%s = %" PRIu32 "\n", "mtu", p->mtu);
+			fprintf(f, "%s = %" PRIu32 "\n", "metadata_size", p->metadata_size);
+			fprintf(f, "%s = %s\n",
+				"mempool_direct",
+				app->mempool_params[p->mempool_direct_id].name);
+			fprintf(f, "%s = %s\n",
+				"mempool_indirect",
+				app->mempool_params[p->mempool_indirect_id].name);
+		}
 
 		fputc('\n', f);
 	}
@@ -2360,15 +2456,31 @@ app_config_init(struct app_params *app)
 	return 0;
 }
 
+static char *
+filenamedup(const char *filename, const char *suffix)
+{
+	char *s = malloc(strlen(filename) + strlen(suffix) + 1);
+
+	if (!s)
+		return NULL;
+
+	sprintf(s, "%s%s", filename, suffix);
+	return s;
+}
+
 int
 app_config_args(struct app_params *app, int argc, char **argv)
 {
-	int opt;
-	int option_index, f_present, s_present, p_present, l_present;
+	const char *optname;
+	int opt, option_index;
+	int f_present, s_present, p_present, l_present;
+	int preproc_present, preproc_params_present;
 	int scaned = 0;
 
 	static struct option lgopts[] = {
-		{NULL, 0, 0, 0}
+		{ "preproc", 1, 0, 0 },
+		{ "preproc-args", 1, 0, 0 },
+		{ NULL,  0, 0, 0 }
 	};
 
 	/* Copy application name */
@@ -2378,6 +2490,8 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	s_present = 0;
 	p_present = 0;
 	l_present = 0;
+	preproc_present = 0;
+	preproc_params_present = 0;
 
 	while ((opt = getopt_long(argc, argv, "f:s:p:l:", lgopts,
 			&option_index)) != EOF)
@@ -2443,6 +2557,32 @@ app_config_args(struct app_params *app, int argc, char **argv)
 
 			break;
 
+		case 0:
+			optname = lgopts[option_index].name;
+
+			if (strcmp(optname, "preproc") == 0) {
+				if (preproc_present)
+					rte_panic("Error: Preprocessor argument "
+						"is provided more than once\n");
+				preproc_present = 1;
+
+				app->preproc = strdup(optarg);
+				break;
+			}
+
+			if (strcmp(optname, "preproc-args") == 0) {
+				if (preproc_params_present)
+					rte_panic("Error: Preprocessor args "
+						"are provided more than once\n");
+				preproc_params_present = 1;
+
+				app->preproc_args = strdup(optarg);
+				break;
+			}
+
+			app_print_usage(argv[0]);
+			break;
+
 		default:
 			app_print_usage(argv[0]);
 		}
@@ -2453,5 +2593,40 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	if (!p_present)
 		rte_panic("Error: PORT_MASK is not provided\n");
 
+	/* Check dependencies between args */
+	if (preproc_params_present && (preproc_present == 0))
+		rte_panic("Error: Preprocessor args specified while "
+			"preprocessor is not defined\n");
+
+	app->parser_file = preproc_present ?
+		filenamedup(app->config_file, ".preproc") :
+		strdup(app->config_file);
+	app->output_file = filenamedup(app->config_file, ".out");
+
 	return 0;
 }
+
+int
+app_config_preproc(struct app_params *app)
+{
+	char buffer[256];
+	int status;
+
+	if (app->preproc == NULL)
+		return 0;
+
+	status = access(app->config_file, F_OK | R_OK);
+	APP_CHECK((status == 0), "Unable to open file %s", app->config_file);
+
+	snprintf(buffer, sizeof(buffer), "%s %s %s > %s",
+		app->preproc,
+		app->preproc_args ? app->preproc_args : "",
+		app->config_file,
+		app->parser_file);
+
+	status = system(buffer);
+	APP_CHECK((WIFEXITED(status) && (WEXITSTATUS(status) == 0)),
+		"Error while preprocessing file \"%s\"\n", app->config_file);
+
+	return status;
+}
diff --git a/examples/ip_pipeline/init.c b/examples/ip_pipeline/init.c
index 3f9c68d..46d044e 100644
--- a/examples/ip_pipeline/init.c
+++ b/examples/ip_pipeline/init.c
@@ -803,11 +803,43 @@ app_check_link(struct app_params *app)
 		rte_panic("Some links are DOWN\n");
 }
 
+static uint32_t
+is_any_swq_frag_or_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	for (i = 0; i < app->n_pktq_swq; i++) {
+		struct app_pktq_swq_params *p = &app->swq_params[i];
+
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1) ||
+			(p->ipv4_ras == 1) || (p->ipv6_ras == 1))
+			return 1;
+	}
+
+	return 0;
+}
+
+static void
+app_init_link_frag_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	if (is_any_swq_frag_or_ras(app)) {
+		for (i = 0; i < app->n_pktq_hwq_out; i++) {
+			struct app_pktq_hwq_out_params *p_txq = &app->hwq_out_params[i];
+
+			p_txq->conf.txq_flags &= ~ETH_TXQ_FLAGS_NOMULTSEGS;
+		}
+	}
+}
+
 static void
 app_init_link(struct app_params *app)
 {
 	uint32_t i;
 
+	app_init_link_frag_ras(app);
+
 	for (i = 0; i < app->n_links; i++) {
 		struct app_link_params *p_link = &app->link_params[i];
 		uint32_t link_id, n_hwq_in, n_hwq_out, j;
@@ -916,13 +948,19 @@ app_init_swq(struct app_params *app)
 
 	for (i = 0; i < app->n_pktq_swq; i++) {
 		struct app_pktq_swq_params *p = &app->swq_params[i];
+		unsigned flags = 0;
+
+		if (app_swq_get_readers(app, p) == 1)
+			flags |= RING_F_SC_DEQ;
+		if (app_swq_get_writers(app, p) == 1)
+			flags |= RING_F_SP_ENQ;
 
 		APP_LOG(app, HIGH, "Initializing %s...", p->name);
 		app->swq[i] = rte_ring_create(
 				p->name,
 				p->size,
 				p->cpu_socket_id,
-				RING_F_SP_ENQ | RING_F_SC_DEQ);
+				flags);
 
 		if (app->swq[i] == NULL)
 			rte_panic("%s init error\n", p->name);
@@ -1059,11 +1097,50 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_IN_SWQ:
-			out->type = PIPELINE_PORT_IN_RING_READER;
-			out->params.ring.ring = app->swq[in->id];
-			out->burst_size = app->swq_params[in->id].burst_read;
-			/* What about frag and ras ports? */
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_frag == 0) && (swq_params->ipv6_frag == 0)) {
+				if (app_swq_get_readers(app, swq_params) == 1) {
+					out->type = PIPELINE_PORT_IN_RING_READER;
+					out->params.ring.ring = app->swq[in->id];
+					out->burst_size = app->swq_params[in->id].burst_read;
+				} else {
+					out->type = PIPELINE_PORT_IN_RING_MULTI_READER;
+					out->params.ring_multi.ring = app->swq[in->id];
+					out->burst_size = swq_params->burst_read;
+				}
+			} else {
+				if (swq_params->ipv4_frag == 1) {
+					struct rte_port_ring_reader_ipv4_frag_params *params =
+						&out->params.ring_ipv4_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV4_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				} else {
+					struct rte_port_ring_reader_ipv6_frag_params *params =
+						&out->params.ring_ipv6_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV6_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				}
+			}
 			break;
+		}
 		case APP_PKTQ_IN_TM:
 			out->type = PIPELINE_PORT_IN_SCHED_READER;
 			out->params.sched.sched = app->tm[in->id];
@@ -1122,28 +1199,68 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_OUT_SWQ:
-			if (app->swq_params[in->id].dropless == 0) {
-				struct rte_port_ring_writer_params *params =
-					&out->params.ring;
-
-				out->type = PIPELINE_PORT_OUT_RING_WRITER;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_ras == 0) && (swq_params->ipv6_ras == 0)) {
+				if (app_swq_get_writers(app, swq_params) == 1) {
+					if (app->swq_params[in->id].dropless == 0) {
+						struct rte_port_ring_writer_params *params =
+							&out->params.ring;
+
+						out->type = PIPELINE_PORT_OUT_RING_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+					} else {
+						struct rte_port_ring_writer_nodrop_params
+							*params = &out->params.ring_nodrop;
+
+						out->type =
+							PIPELINE_PORT_OUT_RING_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+						params->n_retries =
+							app->swq_params[in->id].n_retries;
+					}
+				} else {
+					if (swq_params->dropless == 0) {
+						struct rte_port_ring_multi_writer_params *params =
+							&out->params.ring_multi;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+					} else {
+						struct rte_port_ring_multi_writer_nodrop_params
+							*params = &out->params.ring_multi_nodrop;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+						params->n_retries = swq_params->n_retries;
+					}
+				}
 			} else {
-				struct rte_port_ring_writer_nodrop_params
-					*params = &out->params.ring_nodrop;
-
-				out->type =
-					PIPELINE_PORT_OUT_RING_WRITER_NODROP;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
-				params->n_retries =
-					app->swq_params[in->id].n_retries;
+				if (swq_params->ipv4_ras == 1) {
+					struct rte_port_ring_writer_ipv4_ras_params *params =
+						&out->params.ring_ipv4_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				} else {
+					struct rte_port_ring_writer_ipv6_ras_params *params =
+						&out->params.ring_ipv6_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				}
 			}
-			/* What about frag and ras ports? */
 			break;
+		}
 		case APP_PKTQ_OUT_TM: {
 			struct rte_port_sched_writer_params *params =
 				&out->params.sched;
diff --git a/examples/ip_pipeline/main.c b/examples/ip_pipeline/main.c
index 862e2f2..4944dcf 100644
--- a/examples/ip_pipeline/main.c
+++ b/examples/ip_pipeline/main.c
@@ -45,7 +45,9 @@ main(int argc, char **argv)
 
 	app_config_args(&app, argc, argv);
 
-	app_config_parse(&app, app.config_file);
+	app_config_preproc(&app);
+
+	app_config_parse(&app, app.parser_file);
 
 	app_config_check(&app);
 
diff --git a/examples/ip_pipeline/pipeline_be.h b/examples/ip_pipeline/pipeline_be.h
index 51f1e4f..f7269c0 100644
--- a/examples/ip_pipeline/pipeline_be.h
+++ b/examples/ip_pipeline/pipeline_be.h
@@ -45,6 +45,7 @@
 enum pipeline_port_in_type {
 	PIPELINE_PORT_IN_ETHDEV_READER,
 	PIPELINE_PORT_IN_RING_READER,
+	PIPELINE_PORT_IN_RING_MULTI_READER,
 	PIPELINE_PORT_IN_RING_READER_IPV4_FRAG,
 	PIPELINE_PORT_IN_RING_READER_IPV6_FRAG,
 	PIPELINE_PORT_IN_SCHED_READER,
@@ -56,6 +57,7 @@ struct pipeline_port_in_params {
 	union {
 		struct rte_port_ethdev_reader_params ethdev;
 		struct rte_port_ring_reader_params ring;
+		struct rte_port_ring_multi_reader_params ring_multi;
 		struct rte_port_ring_reader_ipv4_frag_params ring_ipv4_frag;
 		struct rte_port_ring_reader_ipv6_frag_params ring_ipv6_frag;
 		struct rte_port_sched_reader_params sched;
@@ -72,6 +74,8 @@ pipeline_port_in_params_convert(struct pipeline_port_in_params  *p)
 		return (void *) &p->params.ethdev;
 	case PIPELINE_PORT_IN_RING_READER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return (void *) &p->params.ring_ipv4_frag;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -93,6 +97,8 @@ pipeline_port_in_params_get_ops(struct pipeline_port_in_params  *p)
 		return &rte_port_ethdev_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER:
 		return &rte_port_ring_reader_ops;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return &rte_port_ring_multi_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return &rte_port_ring_reader_ipv4_frag_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -110,7 +116,9 @@ enum pipeline_port_out_type {
 	PIPELINE_PORT_OUT_ETHDEV_WRITER,
 	PIPELINE_PORT_OUT_ETHDEV_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER,
 	PIPELINE_PORT_OUT_RING_WRITER_NODROP,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS,
 	PIPELINE_PORT_OUT_SCHED_WRITER,
@@ -123,7 +131,9 @@ struct pipeline_port_out_params {
 		struct rte_port_ethdev_writer_params ethdev;
 		struct rte_port_ethdev_writer_nodrop_params ethdev_nodrop;
 		struct rte_port_ring_writer_params ring;
+		struct rte_port_ring_multi_writer_params ring_multi;
 		struct rte_port_ring_writer_nodrop_params ring_nodrop;
+		struct rte_port_ring_multi_writer_nodrop_params ring_multi_nodrop;
 		struct rte_port_ring_writer_ipv4_ras_params ring_ipv4_ras;
 		struct rte_port_ring_writer_ipv6_ras_params ring_ipv6_ras;
 		struct rte_port_sched_writer_params sched;
@@ -140,8 +150,12 @@ pipeline_port_out_params_convert(struct pipeline_port_out_params  *p)
 		return (void *) &p->params.ethdev_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return (void *) &p->params.ring_nodrop;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return (void *) &p->params.ring_multi_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return (void *) &p->params.ring_ipv4_ras;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
@@ -164,8 +178,12 @@ pipeline_port_out_params_get_ops(struct pipeline_port_out_params  *p)
 		return &rte_port_ethdev_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return &rte_port_ring_writer_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return &rte_port_ring_multi_writer_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return &rte_port_ring_writer_nodrop_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return &rte_port_ring_multi_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return &rte_port_ring_writer_ipv4_ras_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
-- 
1.7.9.5

[dpdk-dev] [PATCH v4 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[Piotr Azarewicz]

This patch set enhancement ip_pipeline application:
- librte_port: add support for multi-producer/multi-consumer ring ports
- librte_port: bug fixes for ring ports with IPv4/IPv6 reassembly support
- ip_pipeline application: integrate MP/MC and fragmentation/reassembly support to SWQs

v2 changes:
- rte_port_ring:
	- fixed checkpatch errors
	- interlace the implementation of multi into the implementation of single
	- reduced the amount of code duplication

v3 changes:
- new functions added in the .map
- add a "Fixes:" tag in commit comment

v4 changes:
- fix usage RTE_MBUF_METADATA_* macros

Acked-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>

Piotr Azarewicz (3):
  port: add mp/mc ring ports
  port: fix ras/frag ring ports
  examples/ip_pipeline: add mp/mc and frag/ras swq

 examples/ip_pipeline/app.h           |   14 ++
 examples/ip_pipeline/config_check.c  |   45 ++++-
 examples/ip_pipeline/config_parse.c  |  195 +++++++++++++++++++--
 examples/ip_pipeline/init.c          |  165 +++++++++++++++---
 examples/ip_pipeline/main.c          |    4 +-
 examples/ip_pipeline/pipeline_be.h   |   18 ++
 lib/librte_port/rte_port_frag.c      |    5 +-
 lib/librte_port/rte_port_ras.c       |    8 +-
 lib/librte_port/rte_port_ring.c      |  311 +++++++++++++++++++++++++++++++---
 lib/librte_port/rte_port_ring.h      |   35 +++-
 lib/librte_port/rte_port_version.map |    9 +
 11 files changed, 736 insertions(+), 73 deletions(-)

-- 
1.7.9.5

[dpdk-dev] [PATCH v4 1/3] port: add mp/mc ring ports

[Piotr Azarewicz]

ring_multi_reader input port (on top of multi consumer rte_ring)
ring_multi_writer output port (on top of multi producer rte_ring)

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_ring.c      |  311 +++++++++++++++++++++++++++++++---
 lib/librte_port/rte_port_ring.h      |   35 +++-
 lib/librte_port/rte_port_version.map |    9 +
 3 files changed, 325 insertions(+), 30 deletions(-)

diff --git a/lib/librte_port/rte_port_ring.c b/lib/librte_port/rte_port_ring.c
index 9461c05..755dfc1 100644
--- a/lib/librte_port/rte_port_ring.c
+++ b/lib/librte_port/rte_port_ring.c
@@ -63,15 +63,19 @@ struct rte_port_ring_reader {
 };
 
 static void *
-rte_port_ring_reader_create(void *params, int socket_id)
+rte_port_ring_reader_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_reader_params *conf =
 			(struct rte_port_ring_reader_params *) params;
 	struct rte_port_ring_reader *port;
 
 	/* Check input parameters */
-	if (conf == NULL) {
-		RTE_LOG(ERR, PORT, "%s: params is NULL\n", __func__);
+	if ((conf == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->cons.sc_dequeue && is_multi) ||
+		(!(conf->ring->cons.sc_dequeue) && !is_multi)) {
+		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
 	}
 
@@ -89,6 +93,18 @@ rte_port_ring_reader_create(void *params, int socket_id)
 	return port;
 }
 
+static void *
+rte_port_ring_reader_create(void *params, int socket_id)
+{
+	return rte_port_ring_reader_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_reader_create(void *params, int socket_id)
+{
+	return rte_port_ring_reader_create_internal(params, socket_id, 1);
+}
+
 static int
 rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
 {
@@ -102,6 +118,19 @@ rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
 }
 
 static int
+rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts,
+	uint32_t n_pkts)
+{
+	struct rte_port_ring_reader *p = (struct rte_port_ring_reader *) port;
+	uint32_t nb_rx;
+
+	nb_rx = rte_ring_mc_dequeue_burst(p->ring, (void **) pkts, n_pkts);
+	RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx);
+
+	return nb_rx;
+}
+
+static int
 rte_port_ring_reader_free(void *port)
 {
 	if (port == NULL) {
@@ -155,10 +184,12 @@ struct rte_port_ring_writer {
 	uint32_t tx_burst_sz;
 	uint32_t tx_buf_count;
 	uint64_t bsz_mask;
+	uint32_t is_multi;
 };
 
 static void *
-rte_port_ring_writer_create(void *params, int socket_id)
+rte_port_ring_writer_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_writer_params *conf =
 			(struct rte_port_ring_writer_params *) params;
@@ -166,7 +197,9 @@ rte_port_ring_writer_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue && is_multi) ||
+		(!(conf->ring->prod.sp_enqueue) && !is_multi) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -185,10 +218,23 @@ rte_port_ring_writer_create(void *params, int socket_id)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+	port->is_multi = is_multi;
 
 	return port;
 }
 
+static void *
+rte_port_ring_writer_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_writer_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_create_internal(params, socket_id, 1);
+}
+
 static inline void
 send_burst(struct rte_port_ring_writer *p)
 {
@@ -204,6 +250,21 @@ send_burst(struct rte_port_ring_writer *p)
 	p->tx_buf_count = 0;
 }
 
+static inline void
+send_burst_mp(struct rte_port_ring_writer *p)
+{
+	uint32_t nb_tx;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+			p->tx_buf_count);
+
+	RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
 static int
 rte_port_ring_writer_tx(void *port, struct rte_mbuf *pkt)
 {
@@ -218,9 +279,23 @@ rte_port_ring_writer_tx(void *port, struct rte_mbuf *pkt)
 }
 
 static int
-rte_port_ring_writer_tx_bulk(void *port,
+rte_port_ring_multi_writer_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static inline int __attribute__((always_inline))
+rte_port_ring_writer_tx_bulk_internal(void *port,
 		struct rte_mbuf **pkts,
-		uint64_t pkts_mask)
+		uint64_t pkts_mask,
+		uint32_t is_multi)
 {
 	struct rte_port_ring_writer *p =
 		(struct rte_port_ring_writer *) port;
@@ -234,11 +309,20 @@ rte_port_ring_writer_tx_bulk(void *port,
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t n_pkts_ok;
 
-		if (tx_buf_count)
-			send_burst(p);
+		if (tx_buf_count) {
+			if (is_multi)
+				send_burst_mp(p);
+			else
+				send_burst(p);
+		}
 
 		RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, n_pkts);
-		n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		if (is_multi)
+			n_pkts_ok = rte_ring_mp_enqueue_burst(p->ring, (void **)pkts,
+				n_pkts);
+		else
+			n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts,
+				n_pkts);
 
 		RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, n_pkts - n_pkts_ok);
 		for ( ; n_pkts_ok < n_pkts; n_pkts_ok++) {
@@ -258,14 +342,34 @@ rte_port_ring_writer_tx_bulk(void *port,
 		}
 
 		p->tx_buf_count = tx_buf_count;
-		if (tx_buf_count >= p->tx_burst_sz)
-			send_burst(p);
+		if (tx_buf_count >= p->tx_burst_sz) {
+			if (is_multi)
+				send_burst_mp(p);
+			else
+				send_burst(p);
+		}
 	}
 
 	return 0;
 }
 
 static int
+rte_port_ring_writer_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 0);
+}
+
+static int
+rte_port_ring_multi_writer_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 1);
+}
+
+static int
 rte_port_ring_writer_flush(void *port)
 {
 	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
@@ -277,14 +381,31 @@ rte_port_ring_writer_flush(void *port)
 }
 
 static int
+rte_port_ring_multi_writer_flush(void *port)
+{
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp(p);
+
+	return 0;
+}
+
+static int
 rte_port_ring_writer_free(void *port)
 {
+	struct rte_port_ring_writer *p = (struct rte_port_ring_writer *) port;
+
 	if (port == NULL) {
 		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
 		return -EINVAL;
 	}
 
-	rte_port_ring_writer_flush(port);
+	if (p->is_multi)
+		rte_port_ring_multi_writer_flush(port);
+	else
+		rte_port_ring_writer_flush(port);
+
 	rte_free(port);
 
 	return 0;
@@ -332,10 +453,12 @@ struct rte_port_ring_writer_nodrop {
 	uint32_t tx_buf_count;
 	uint64_t bsz_mask;
 	uint64_t n_retries;
+	uint32_t is_multi;
 };
 
 static void *
-rte_port_ring_writer_nodrop_create(void *params, int socket_id)
+rte_port_ring_writer_nodrop_create_internal(void *params, int socket_id,
+	uint32_t is_multi)
 {
 	struct rte_port_ring_writer_nodrop_params *conf =
 			(struct rte_port_ring_writer_nodrop_params *) params;
@@ -343,7 +466,9 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 
 	/* Check input parameters */
 	if ((conf == NULL) ||
-	    (conf->ring == NULL) ||
+		(conf->ring == NULL) ||
+		(conf->ring->prod.sp_enqueue && is_multi) ||
+		(!(conf->ring->prod.sp_enqueue) && !is_multi) ||
 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
 		RTE_LOG(ERR, PORT, "%s: Invalid Parameters\n", __func__);
 		return NULL;
@@ -362,6 +487,7 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
+	port->is_multi = is_multi;
 
 	/*
 	 * When n_retries is 0 it means that we should wait for every packet to
@@ -373,6 +499,18 @@ rte_port_ring_writer_nodrop_create(void *params, int socket_id)
 	return port;
 }
 
+static void *
+rte_port_ring_writer_nodrop_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 0);
+}
+
+static void *
+rte_port_ring_multi_writer_nodrop_create(void *params, int socket_id)
+{
+	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 1);
+}
+
 static inline void
 send_burst_nodrop(struct rte_port_ring_writer_nodrop *p)
 {
@@ -402,6 +540,35 @@ send_burst_nodrop(struct rte_port_ring_writer_nodrop *p)
 	p->tx_buf_count = 0;
 }
 
+static inline void
+send_burst_mp_nodrop(struct rte_port_ring_writer_nodrop *p)
+{
+	uint32_t nb_tx = 0, i;
+
+	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
+				p->tx_buf_count);
+
+	/* We sent all the packets in a first try */
+	if (nb_tx >= p->tx_buf_count)
+		return;
+
+	for (i = 0; i < p->n_retries; i++) {
+		nb_tx += rte_ring_mp_enqueue_burst(p->ring,
+				(void **) (p->tx_buf + nb_tx), p->tx_buf_count - nb_tx);
+
+		/* We sent all the packets in more than one try */
+		if (nb_tx >= p->tx_buf_count)
+			return;
+	}
+
+	/* We didn't send the packets in maximum allowed attempts */
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
+	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
+		rte_pktmbuf_free(p->tx_buf[nb_tx]);
+
+	p->tx_buf_count = 0;
+}
+
 static int
 rte_port_ring_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
 {
@@ -417,9 +584,24 @@ rte_port_ring_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
 }
 
 static int
-rte_port_ring_writer_nodrop_tx_bulk(void *port,
+rte_port_ring_multi_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	p->tx_buf[p->tx_buf_count++] = pkt;
+	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
+	if (p->tx_buf_count >= p->tx_burst_sz)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static inline int __attribute__((always_inline))
+rte_port_ring_writer_nodrop_tx_bulk_internal(void *port,
 		struct rte_mbuf **pkts,
-		uint64_t pkts_mask)
+		uint64_t pkts_mask,
+		uint32_t is_multi)
 {
 	struct rte_port_ring_writer_nodrop *p =
 		(struct rte_port_ring_writer_nodrop *) port;
@@ -433,24 +615,37 @@ rte_port_ring_writer_nodrop_tx_bulk(void *port,
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t n_pkts_ok;
 
-		if (tx_buf_count)
-			send_burst_nodrop(p);
+		if (tx_buf_count) {
+			if (is_multi)
+				send_burst_mp_nodrop(p);
+			else
+				send_burst_nodrop(p);
+		}
 
 		RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, n_pkts);
-		n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		if (is_multi)
+			n_pkts_ok =
+				rte_ring_mp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
+		else
+			n_pkts_ok =
+				rte_ring_sp_enqueue_burst(p->ring, (void **)pkts, n_pkts);
 
 		if (n_pkts_ok >= n_pkts)
 			return 0;
 
 		/*
-		 * If we didnt manage to send all packets in single burst, move
+		 * If we didn't manage to send all packets in single burst, move
 		 * remaining packets to the buffer and call send burst.
 		 */
 		for (; n_pkts_ok < n_pkts; n_pkts_ok++) {
 			struct rte_mbuf *pkt = pkts[n_pkts_ok];
+
 			p->tx_buf[p->tx_buf_count++] = pkt;
 		}
-		send_burst_nodrop(p);
+		if (is_multi)
+			send_burst_mp_nodrop(p);
+		else
+			send_burst_nodrop(p);
 	} else {
 		for ( ; pkts_mask; ) {
 			uint32_t pkt_index = __builtin_ctzll(pkts_mask);
@@ -463,14 +658,36 @@ rte_port_ring_writer_nodrop_tx_bulk(void *port,
 		}
 
 		p->tx_buf_count = tx_buf_count;
-		if (tx_buf_count >= p->tx_burst_sz)
-			send_burst_nodrop(p);
+		if (tx_buf_count >= p->tx_burst_sz) {
+			if (is_multi)
+				send_burst_mp_nodrop(p);
+			else
+				send_burst_nodrop(p);
+		}
 	}
 
 	return 0;
 }
 
 static int
+rte_port_ring_writer_nodrop_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return
+		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 0);
+}
+
+static int
+rte_port_ring_multi_writer_nodrop_tx_bulk(void *port,
+		struct rte_mbuf **pkts,
+		uint64_t pkts_mask)
+{
+	return
+		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 1);
+}
+
+static int
 rte_port_ring_writer_nodrop_flush(void *port)
 {
 	struct rte_port_ring_writer_nodrop *p =
@@ -483,14 +700,33 @@ rte_port_ring_writer_nodrop_flush(void *port)
 }
 
 static int
+rte_port_ring_multi_writer_nodrop_flush(void *port)
+{
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
+	if (p->tx_buf_count > 0)
+		send_burst_mp_nodrop(p);
+
+	return 0;
+}
+
+static int
 rte_port_ring_writer_nodrop_free(void *port)
 {
+	struct rte_port_ring_writer_nodrop *p =
+			(struct rte_port_ring_writer_nodrop *) port;
+
 	if (port == NULL) {
 		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
 		return -EINVAL;
 	}
 
-	rte_port_ring_writer_nodrop_flush(port);
+	if (p->is_multi)
+		rte_port_ring_multi_writer_nodrop_flush(port);
+	else
+		rte_port_ring_writer_nodrop_flush(port);
+
 	rte_free(port);
 
 	return 0;
@@ -539,3 +775,28 @@ struct rte_port_out_ops rte_port_ring_writer_nodrop_ops = {
 	.f_flush = rte_port_ring_writer_nodrop_flush,
 	.f_stats = rte_port_ring_writer_nodrop_stats_read,
 };
+
+struct rte_port_in_ops rte_port_ring_multi_reader_ops = {
+	.f_create = rte_port_ring_multi_reader_create,
+	.f_free = rte_port_ring_reader_free,
+	.f_rx = rte_port_ring_multi_reader_rx,
+	.f_stats = rte_port_ring_reader_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_ops = {
+	.f_create = rte_port_ring_multi_writer_create,
+	.f_free = rte_port_ring_writer_free,
+	.f_tx = rte_port_ring_multi_writer_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_flush,
+	.f_stats = rte_port_ring_writer_stats_read,
+};
+
+struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops = {
+	.f_create = rte_port_ring_multi_writer_nodrop_create,
+	.f_free = rte_port_ring_writer_nodrop_free,
+	.f_tx = rte_port_ring_multi_writer_nodrop_tx,
+	.f_tx_bulk = rte_port_ring_multi_writer_nodrop_tx_bulk,
+	.f_flush = rte_port_ring_multi_writer_nodrop_flush,
+	.f_stats = rte_port_ring_writer_nodrop_stats_read,
+};
diff --git a/lib/librte_port/rte_port_ring.h b/lib/librte_port/rte_port_ring.h
index 89a219b..de377d2 100644
--- a/lib/librte_port/rte_port_ring.h
+++ b/lib/librte_port/rte_port_ring.h
@@ -42,8 +42,14 @@ extern "C" {
  * @file
  * RTE Port Ring
  *
- * ring_reader: input port built on top of pre-initialized single consumer ring
- * ring_writer: output port built on top of pre-initialized single producer ring
+ * ring_reader:
+ *      input port built on top of pre-initialized single consumer ring
+ * ring_writer:
+ *      output port built on top of pre-initialized single producer ring
+ * ring_multi_reader:
+ *      input port built on top of pre-initialized multi consumers ring
+ * ring_multi_writer:
+ *      output port built on top of pre-initialized multi producers ring
  *
  ***/
 
@@ -55,7 +61,7 @@ extern "C" {
 
 /** ring_reader port parameters */
 struct rte_port_ring_reader_params {
-	/** Underlying single consumer ring that has to be pre-initialized */
+	/** Underlying consumer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 };
 
@@ -64,7 +70,7 @@ extern struct rte_port_in_ops rte_port_ring_reader_ops;
 
 /** ring_writer port parameters */
 struct rte_port_ring_writer_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -77,7 +83,7 @@ extern struct rte_port_out_ops rte_port_ring_writer_ops;
 
 /** ring_writer_nodrop port parameters */
 struct rte_port_ring_writer_nodrop_params {
-	/** Underlying single producer ring that has to be pre-initialized */
+	/** Underlying producer ring that has to be pre-initialized */
 	struct rte_ring *ring;
 
 	/** Recommended burst size to ring. The actual burst size can be
@@ -91,6 +97,25 @@ struct rte_port_ring_writer_nodrop_params {
 /** ring_writer_nodrop port operations */
 extern struct rte_port_out_ops rte_port_ring_writer_nodrop_ops;
 
+/** ring_multi_reader port parameters */
+#define rte_port_ring_multi_reader_params rte_port_ring_reader_params
+
+/** ring_multi_reader port operations */
+extern struct rte_port_in_ops rte_port_ring_multi_reader_ops;
+
+/** ring_multi_writer port parameters */
+#define rte_port_ring_multi_writer_params rte_port_ring_writer_params
+
+/** ring_multi_writer port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_ops;
+
+/** ring_multi_writer_nodrop port parameters */
+#define rte_port_ring_multi_writer_nodrop_params \
+	rte_port_ring_writer_nodrop_params
+
+/** ring_multi_writer_nodrop port operations */
+extern struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops;
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/librte_port/rte_port_version.map b/lib/librte_port/rte_port_version.map
index 4e49ff6..7a0b34d 100644
--- a/lib/librte_port/rte_port_version.map
+++ b/lib/librte_port/rte_port_version.map
@@ -26,3 +26,12 @@ DPDK_2.1 {
 	rte_port_ring_writer_nodrop_ops;
 
 } DPDK_2.0;
+
+DPDK_2.2 {
+	global:
+
+	rte_port_ring_multi_reader_ops;
+	rte_port_ring_multi_writer_ops;
+	rte_port_ring_multi_writer_nodrop_ops;
+
+} DPDK_2.1;
-- 
1.7.9.5

[dpdk-dev] [PATCH v4 2/3] port: fix ras/frag ring ports

[Piotr Azarewicz]

Bug fixes for ring ports with IPv4/IPv6 reassembly support.
Previous implementation can't work properly due to incorrect choosing
process function.
Also, assuming that, when processing ip packet, ip header is know we can
set l3_len parameter here.

Fix usage RTE_MBUF_METADATA_* macros due to redefinition the macros.

Fixes: 50f54a84dfb7 ("port: add IPv6 reassembly port")
Fixes: ba92d511ddac ("port: move metadata offset reference at mbuf head")

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 lib/librte_port/rte_port_frag.c |    5 +++--
 lib/librte_port/rte_port_ras.c  |    8 ++++++--
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/lib/librte_port/rte_port_frag.c b/lib/librte_port/rte_port_frag.c
index 3720d5d..0fcace9 100644
--- a/lib/librte_port/rte_port_frag.c
+++ b/lib/librte_port/rte_port_frag.c
@@ -229,9 +229,10 @@ rte_port_ring_reader_frag_rx(void *port,
 
 		/* Copy meta-data from input jumbo packet to its fragments */
 		for (i = 0; i < p->n_frags; i++) {
-			uint8_t *src = RTE_MBUF_METADATA_UINT8_PTR(pkt, 0);
+			uint8_t *src =
+			  RTE_MBUF_METADATA_UINT8_PTR(pkt, sizeof(struct rte_mbuf));
 			uint8_t *dst =
-				RTE_MBUF_METADATA_UINT8_PTR(p->frags[i], 0);
+			  RTE_MBUF_METADATA_UINT8_PTR(p->frags[i], sizeof(struct rte_mbuf));
 
 			memcpy(dst, src, p->metadata_size);
 		}
diff --git a/lib/librte_port/rte_port_ras.c b/lib/librte_port/rte_port_ras.c
index 8a2e554..c4bb508 100644
--- a/lib/librte_port/rte_port_ras.c
+++ b/lib/librte_port/rte_port_ras.c
@@ -144,7 +144,7 @@ rte_port_ring_writer_ras_create(void *params, int socket_id, int is_ipv4)
 	port->tx_burst_sz = conf->tx_burst_sz;
 	port->tx_buf_count = 0;
 
-	port->f_ras = (is_ipv4 == 0) ? process_ipv4 : process_ipv6;
+	port->f_ras = (is_ipv4 == 1) ? process_ipv4 : process_ipv6;
 
 	return port;
 }
@@ -182,7 +182,7 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 	/* Assume there is no ethernet header */
 	struct ipv4_hdr *pkt_hdr = rte_pktmbuf_mtod(pkt, struct ipv4_hdr *);
 
-	/* Get "Do not fragment" flag and fragment offset */
+	/* Get "More fragments" flag and fragment offset */
 	uint16_t frag_field = rte_be_to_cpu_16(pkt_hdr->fragment_offset);
 	uint16_t frag_offset = (uint16_t)(frag_field & IPV4_HDR_OFFSET_MASK);
 	uint16_t frag_flag = (uint16_t)(frag_field & IPV4_HDR_MF_FLAG);
@@ -195,6 +195,8 @@ process_ipv4(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv4_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(),
 				pkt_hdr);
@@ -225,6 +227,8 @@ process_ipv6(struct rte_port_ring_writer_ras *p, struct rte_mbuf *pkt)
 		struct rte_ip_frag_tbl *tbl = p->frag_tbl;
 		struct rte_ip_frag_death_row *dr = &p->death_row;
 
+		pkt->l3_len = sizeof(*pkt_hdr) + sizeof(*frag_hdr);
+
 		/* Process this fragment */
 		mo = rte_ipv6_frag_reassemble_packet(tbl, dr, pkt, rte_rdtsc(), pkt_hdr,
 				frag_hdr);
-- 
1.7.9.5

[dpdk-dev] [PATCH v4 3/3] examples/ip_pipeline: add mp/mc and frag/ras swq

[Piotr Azarewicz]

Add integrated MP/MC and fragmentation/reassembly support to SWQs

Signed-off-by: Piotr Azarewicz <piotrx.t.azarewicz@intel.com>
---
 examples/ip_pipeline/app.h          |   14 +++
 examples/ip_pipeline/config_check.c |   45 +++++++-
 examples/ip_pipeline/config_parse.c |  195 +++++++++++++++++++++++++++++++++--
 examples/ip_pipeline/init.c         |  165 ++++++++++++++++++++++++-----
 examples/ip_pipeline/main.c         |    4 +-
 examples/ip_pipeline/pipeline_be.h  |   18 ++++
 6 files changed, 402 insertions(+), 39 deletions(-)

diff --git a/examples/ip_pipeline/app.h b/examples/ip_pipeline/app.h
index 521e3a0..943466e 100644
--- a/examples/ip_pipeline/app.h
+++ b/examples/ip_pipeline/app.h
@@ -107,6 +107,14 @@ struct app_pktq_swq_params {
 	uint32_t dropless;
 	uint64_t n_retries;
 	uint32_t cpu_socket_id;
+	uint32_t ipv4_frag;
+	uint32_t ipv6_frag;
+	uint32_t ipv4_ras;
+	uint32_t ipv6_ras;
+	uint32_t mtu;
+	uint32_t metadata_size;
+	uint32_t mempool_direct_id;
+	uint32_t mempool_indirect_id;
 };
 
 #ifndef APP_FILE_NAME_SIZE
@@ -405,6 +413,10 @@ struct app_params {
 	char app_name[APP_APPNAME_SIZE];
 	const char *config_file;
 	const char *script_file;
+	const char *parser_file;
+	const char *output_file;
+	const char *preproc;
+	const char *preproc_args;
 	uint64_t port_mask;
 	uint32_t log_level;
 
@@ -880,6 +892,8 @@ int app_config_init(struct app_params *app);
 int app_config_args(struct app_params *app,
 	int argc, char **argv);
 
+int app_config_preproc(struct app_params *app);
+
 int app_config_parse(struct app_params *app,
 	const char *file_name);
 
diff --git a/examples/ip_pipeline/config_check.c b/examples/ip_pipeline/config_check.c
index 07f4c8b..8052bc4 100644
--- a/examples/ip_pipeline/config_check.c
+++ b/examples/ip_pipeline/config_check.c
@@ -33,6 +33,8 @@
 
 #include <stdio.h>
 
+#include <rte_ip.h>
+
 #include "app.h"
 
 static void
@@ -193,6 +195,7 @@ check_swqs(struct app_params *app)
 		struct app_pktq_swq_params *p = &app->swq_params[i];
 		uint32_t n_readers = app_swq_get_readers(app, p);
 		uint32_t n_writers = app_swq_get_writers(app, p);
+		uint32_t n_flags;
 
 		APP_CHECK((p->size > 0),
 			"%s size is 0\n", p->name);
@@ -217,14 +220,48 @@ check_swqs(struct app_params *app)
 		APP_CHECK((n_readers != 0),
 			"%s has no reader\n", p->name);
 
-		APP_CHECK((n_readers == 1),
-			"%s has more than one reader\n", p->name);
+		if (n_readers > 1)
+			APP_LOG(app, LOW, "%s has more than one reader", p->name);
 
 		APP_CHECK((n_writers != 0),
 			"%s has no writer\n", p->name);
 
-		APP_CHECK((n_writers == 1),
-			"%s has more than one writer\n", p->name);
+		if (n_writers > 1)
+			APP_LOG(app, LOW, "%s has more than one writer", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag + p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((n_flags < 2),
+			"%s has more than one fragmentation or reassembly mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_readers > 1) && (n_flags == 1))),
+			"%s has more than one reader when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		APP_CHECK((!((n_writers > 1) && (n_flags == 1))),
+			"%s has more than one writer when fragmentation or reassembly"
+			" mode enabled\n",
+			p->name);
+
+		n_flags = p->ipv4_ras + p->ipv6_ras;
+
+		APP_CHECK((!((p->dropless == 1) && (n_flags == 1))),
+			"%s has dropless when reassembly mode enabled\n", p->name);
+
+		n_flags = p->ipv4_frag + p->ipv6_frag;
+
+		if (n_flags == 1) {
+			uint16_t ip_hdr_size = (p->ipv4_frag) ? sizeof(struct ipv4_hdr) :
+				sizeof(struct ipv6_hdr);
+
+			APP_CHECK((p->mtu > ip_hdr_size),
+				"%s mtu size is smaller than ip header\n", p->name);
+
+			APP_CHECK((!((p->mtu - ip_hdr_size) % 8)),
+				"%s mtu size is incorrect\n", p->name);
+		}
 	}
 }
 
diff --git a/examples/ip_pipeline/config_parse.c b/examples/ip_pipeline/config_parse.c
index c9b78f9..a35bd3e 100644
--- a/examples/ip_pipeline/config_parse.c
+++ b/examples/ip_pipeline/config_parse.c
@@ -156,6 +156,14 @@ static const struct app_pktq_swq_params default_swq_params = {
 	.dropless = 0,
 	.n_retries = 0,
 	.cpu_socket_id = 0,
+	.ipv4_frag = 0,
+	.ipv6_frag = 0,
+	.ipv4_ras = 0,
+	.ipv6_ras = 0,
+	.mtu = 0,
+	.metadata_size = 0,
+	.mempool_direct_id = 0,
+	.mempool_indirect_id = 0,
 };
 
 struct app_pktq_tm_params default_tm_params = {
@@ -196,13 +204,15 @@ struct app_pipeline_params default_pipeline_params = {
 
 static const char app_usage[] =
 	"Usage: %s [-f CONFIG_FILE] [-s SCRIPT_FILE] -p PORT_MASK "
-	"[-l LOG_LEVEL]\n"
+	"[-l LOG_LEVEL] [--preproc PREPROCESSOR] [--preproc-args ARGS]\n"
 	"\n"
 	"Arguments:\n"
 	"\t-f CONFIG_FILE: Default config file is %s\n"
 	"\t-p PORT_MASK: Mask of NIC port IDs in hexadecimal format\n"
 	"\t-s SCRIPT_FILE: No CLI script file is run when not specified\n"
 	"\t-l LOG_LEVEL: 0 = NONE, 1 = HIGH PRIO (default), 2 = LOW PRIO\n"
+	"\t--preproc PREPROCESSOR: Configuration file pre-processor\n"
+	"\t--preproc-args ARGS: Arguments to be passed to pre-processor\n"
 	"\n";
 
 static void
@@ -1107,6 +1117,10 @@ parse_pipeline(struct app_params *app,
 			ret = parser_read_uint32(&param->timer_period,
 				ent->value);
 		else {
+			APP_CHECK((param->n_args < APP_MAX_PIPELINE_ARGS),
+				"CFG: [%s] out of memory",
+				section_name);
+
 			param->args_name[param->n_args] = strdup(ent->name);
 			param->args_value[param->n_args] = strdup(ent->value);
 
@@ -1397,6 +1411,7 @@ parse_swq(struct app_params *app,
 	struct app_pktq_swq_params *param;
 	struct rte_cfgfile_entry *entries;
 	int n_entries, ret, i;
+	unsigned frag_entries = 0;
 	ssize_t param_idx;
 
 	n_entries = rte_cfgfile_section_num_entries(cfg, section_name);
@@ -1438,6 +1453,71 @@ parse_swq(struct app_params *app,
 		else if (strcmp(ent->name, "cpu") == 0)
 			ret = parser_read_uint32(&param->cpu_socket_id,
 				ent->value);
+		else if (strcmp(ent->name, "ipv4_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1500;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_frag") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_frag = ret;
+				if (param->mtu == 0)
+					param->mtu = 1320;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv4_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv4_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "ipv6_ras") == 0) {
+			ret = parser_read_arg_bool(ent->value);
+			if (ret >= 0) {
+				param->ipv6_ras = ret;
+				ret = 0;
+			}
+		} else if (strcmp(ent->name, "mtu") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->mtu,
+				ent->value);
+		} else if (strcmp(ent->name, "metadata_size") == 0) {
+			frag_entries = 1;
+			ret = parser_read_uint32(&param->metadata_size,
+				ent->value);
+		} else if (strcmp(ent->name, "mempool_direct") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_direct_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		} else if (strcmp(ent->name, "mempool_indirect") == 0) {
+			int status = validate_name(ent->value, "MEMPOOL", 1);
+			ssize_t idx;
+
+			APP_CHECK((status == 0),
+				"CFG: [%s] entry '%s': invalid mempool\n",
+				section_name,
+				ent->name);
+
+			idx = APP_PARAM_ADD(app->mempool_params, ent->value);
+			PARSER_IMPLICIT_PARAM_ADD_CHECK(idx, section_name);
+			param->mempool_indirect_id = idx;
+			frag_entries = 1;
+			ret = 0;
+		}
 
 		APP_CHECK(ret != -ESRCH,
 			"CFG: [%s] entry '%s': unknown entry\n",
@@ -1450,6 +1530,13 @@ parse_swq(struct app_params *app,
 			ent->value);
 	}
 
+	if (frag_entries == 1) {
+		APP_CHECK(((param->ipv4_frag == 1) || (param->ipv6_frag == 1)),
+			"CFG: [%s] ipv4/ipv6 frag is off : unsupported entries on this"
+			" configuration\n",
+			section_name);
+	}
+
 	free(entries);
 }
 
@@ -1769,7 +1856,6 @@ parse_port_mask(struct app_params *app, uint64_t port_mask)
 int
 app_config_parse(struct app_params *app, const char *file_name)
 {
-	char config_file_out[APP_FILE_NAME_SIZE];
 	struct rte_cfgfile *cfg;
 	char **section_names;
 	int i, j, sect_count;
@@ -1851,11 +1937,7 @@ app_config_parse(struct app_params *app, const char *file_name)
 	APP_PARAM_COUNT(app->pipeline_params, app->n_pipelines);
 
 	/* Save configuration to output file */
-	snprintf(config_file_out,
-		APP_FILE_NAME_SIZE,
-		"%s.out",
-		app->config_file);
-	app_config_save(app, config_file_out);
+	app_config_save(app, app->output_file);
 
 	/* Load TM configuration files */
 	app_config_parse_tm(app);
@@ -2069,6 +2151,20 @@ save_swq_params(struct app_params *app, FILE *f)
 		fprintf(f, "%s = %s\n", "dropless", p->dropless ? "yes" : "no");
 		fprintf(f, "%s = %" PRIu64 "\n", "n_retries", p->n_retries);
 		fprintf(f, "%s = %" PRIu32 "\n", "cpu", p->cpu_socket_id);
+		fprintf(f, "%s = %s\n", "ipv4_frag", p->ipv4_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_frag", p->ipv6_frag ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv4_ras", p->ipv4_ras ? "yes" : "no");
+		fprintf(f, "%s = %s\n", "ipv6_ras", p->ipv6_ras ? "yes" : "no");
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1)) {
+			fprintf(f, "%s = %" PRIu32 "\n", "mtu", p->mtu);
+			fprintf(f, "%s = %" PRIu32 "\n", "metadata_size", p->metadata_size);
+			fprintf(f, "%s = %s\n",
+				"mempool_direct",
+				app->mempool_params[p->mempool_direct_id].name);
+			fprintf(f, "%s = %s\n",
+				"mempool_indirect",
+				app->mempool_params[p->mempool_indirect_id].name);
+		}
 
 		fputc('\n', f);
 	}
@@ -2360,15 +2456,31 @@ app_config_init(struct app_params *app)
 	return 0;
 }
 
+static char *
+filenamedup(const char *filename, const char *suffix)
+{
+	char *s = malloc(strlen(filename) + strlen(suffix) + 1);
+
+	if (!s)
+		return NULL;
+
+	sprintf(s, "%s%s", filename, suffix);
+	return s;
+}
+
 int
 app_config_args(struct app_params *app, int argc, char **argv)
 {
-	int opt;
-	int option_index, f_present, s_present, p_present, l_present;
+	const char *optname;
+	int opt, option_index;
+	int f_present, s_present, p_present, l_present;
+	int preproc_present, preproc_params_present;
 	int scaned = 0;
 
 	static struct option lgopts[] = {
-		{NULL, 0, 0, 0}
+		{ "preproc", 1, 0, 0 },
+		{ "preproc-args", 1, 0, 0 },
+		{ NULL,  0, 0, 0 }
 	};
 
 	/* Copy application name */
@@ -2378,6 +2490,8 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	s_present = 0;
 	p_present = 0;
 	l_present = 0;
+	preproc_present = 0;
+	preproc_params_present = 0;
 
 	while ((opt = getopt_long(argc, argv, "f:s:p:l:", lgopts,
 			&option_index)) != EOF)
@@ -2443,6 +2557,32 @@ app_config_args(struct app_params *app, int argc, char **argv)
 
 			break;
 
+		case 0:
+			optname = lgopts[option_index].name;
+
+			if (strcmp(optname, "preproc") == 0) {
+				if (preproc_present)
+					rte_panic("Error: Preprocessor argument "
+						"is provided more than once\n");
+				preproc_present = 1;
+
+				app->preproc = strdup(optarg);
+				break;
+			}
+
+			if (strcmp(optname, "preproc-args") == 0) {
+				if (preproc_params_present)
+					rte_panic("Error: Preprocessor args "
+						"are provided more than once\n");
+				preproc_params_present = 1;
+
+				app->preproc_args = strdup(optarg);
+				break;
+			}
+
+			app_print_usage(argv[0]);
+			break;
+
 		default:
 			app_print_usage(argv[0]);
 		}
@@ -2453,5 +2593,40 @@ app_config_args(struct app_params *app, int argc, char **argv)
 	if (!p_present)
 		rte_panic("Error: PORT_MASK is not provided\n");
 
+	/* Check dependencies between args */
+	if (preproc_params_present && (preproc_present == 0))
+		rte_panic("Error: Preprocessor args specified while "
+			"preprocessor is not defined\n");
+
+	app->parser_file = preproc_present ?
+		filenamedup(app->config_file, ".preproc") :
+		strdup(app->config_file);
+	app->output_file = filenamedup(app->config_file, ".out");
+
 	return 0;
 }
+
+int
+app_config_preproc(struct app_params *app)
+{
+	char buffer[256];
+	int status;
+
+	if (app->preproc == NULL)
+		return 0;
+
+	status = access(app->config_file, F_OK | R_OK);
+	APP_CHECK((status == 0), "Unable to open file %s", app->config_file);
+
+	snprintf(buffer, sizeof(buffer), "%s %s %s > %s",
+		app->preproc,
+		app->preproc_args ? app->preproc_args : "",
+		app->config_file,
+		app->parser_file);
+
+	status = system(buffer);
+	APP_CHECK((WIFEXITED(status) && (WEXITSTATUS(status) == 0)),
+		"Error while preprocessing file \"%s\"\n", app->config_file);
+
+	return status;
+}
diff --git a/examples/ip_pipeline/init.c b/examples/ip_pipeline/init.c
index 3f9c68d..46d044e 100644
--- a/examples/ip_pipeline/init.c
+++ b/examples/ip_pipeline/init.c
@@ -803,11 +803,43 @@ app_check_link(struct app_params *app)
 		rte_panic("Some links are DOWN\n");
 }
 
+static uint32_t
+is_any_swq_frag_or_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	for (i = 0; i < app->n_pktq_swq; i++) {
+		struct app_pktq_swq_params *p = &app->swq_params[i];
+
+		if ((p->ipv4_frag == 1) || (p->ipv6_frag == 1) ||
+			(p->ipv4_ras == 1) || (p->ipv6_ras == 1))
+			return 1;
+	}
+
+	return 0;
+}
+
+static void
+app_init_link_frag_ras(struct app_params *app)
+{
+	uint32_t i;
+
+	if (is_any_swq_frag_or_ras(app)) {
+		for (i = 0; i < app->n_pktq_hwq_out; i++) {
+			struct app_pktq_hwq_out_params *p_txq = &app->hwq_out_params[i];
+
+			p_txq->conf.txq_flags &= ~ETH_TXQ_FLAGS_NOMULTSEGS;
+		}
+	}
+}
+
 static void
 app_init_link(struct app_params *app)
 {
 	uint32_t i;
 
+	app_init_link_frag_ras(app);
+
 	for (i = 0; i < app->n_links; i++) {
 		struct app_link_params *p_link = &app->link_params[i];
 		uint32_t link_id, n_hwq_in, n_hwq_out, j;
@@ -916,13 +948,19 @@ app_init_swq(struct app_params *app)
 
 	for (i = 0; i < app->n_pktq_swq; i++) {
 		struct app_pktq_swq_params *p = &app->swq_params[i];
+		unsigned flags = 0;
+
+		if (app_swq_get_readers(app, p) == 1)
+			flags |= RING_F_SC_DEQ;
+		if (app_swq_get_writers(app, p) == 1)
+			flags |= RING_F_SP_ENQ;
 
 		APP_LOG(app, HIGH, "Initializing %s...", p->name);
 		app->swq[i] = rte_ring_create(
 				p->name,
 				p->size,
 				p->cpu_socket_id,
-				RING_F_SP_ENQ | RING_F_SC_DEQ);
+				flags);
 
 		if (app->swq[i] == NULL)
 			rte_panic("%s init error\n", p->name);
@@ -1059,11 +1097,50 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_IN_SWQ:
-			out->type = PIPELINE_PORT_IN_RING_READER;
-			out->params.ring.ring = app->swq[in->id];
-			out->burst_size = app->swq_params[in->id].burst_read;
-			/* What about frag and ras ports? */
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_frag == 0) && (swq_params->ipv6_frag == 0)) {
+				if (app_swq_get_readers(app, swq_params) == 1) {
+					out->type = PIPELINE_PORT_IN_RING_READER;
+					out->params.ring.ring = app->swq[in->id];
+					out->burst_size = app->swq_params[in->id].burst_read;
+				} else {
+					out->type = PIPELINE_PORT_IN_RING_MULTI_READER;
+					out->params.ring_multi.ring = app->swq[in->id];
+					out->burst_size = swq_params->burst_read;
+				}
+			} else {
+				if (swq_params->ipv4_frag == 1) {
+					struct rte_port_ring_reader_ipv4_frag_params *params =
+						&out->params.ring_ipv4_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV4_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				} else {
+					struct rte_port_ring_reader_ipv6_frag_params *params =
+						&out->params.ring_ipv6_frag;
+
+					out->type = PIPELINE_PORT_IN_RING_READER_IPV6_FRAG;
+					params->ring = app->swq[in->id];
+					params->mtu = swq_params->mtu;
+					params->metadata_size = swq_params->metadata_size;
+					params->pool_direct =
+						app->mempool[swq_params->mempool_direct_id];
+					params->pool_indirect =
+						app->mempool[swq_params->mempool_indirect_id];
+					out->burst_size = swq_params->burst_read;
+				}
+			}
 			break;
+		}
 		case APP_PKTQ_IN_TM:
 			out->type = PIPELINE_PORT_IN_SCHED_READER;
 			out->params.sched.sched = app->tm[in->id];
@@ -1122,28 +1199,68 @@ static void app_pipeline_params_get(struct app_params *app,
 			break;
 		}
 		case APP_PKTQ_OUT_SWQ:
-			if (app->swq_params[in->id].dropless == 0) {
-				struct rte_port_ring_writer_params *params =
-					&out->params.ring;
-
-				out->type = PIPELINE_PORT_OUT_RING_WRITER;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
+		{
+			struct app_pktq_swq_params *swq_params = &app->swq_params[in->id];
+
+			if ((swq_params->ipv4_ras == 0) && (swq_params->ipv6_ras == 0)) {
+				if (app_swq_get_writers(app, swq_params) == 1) {
+					if (app->swq_params[in->id].dropless == 0) {
+						struct rte_port_ring_writer_params *params =
+							&out->params.ring;
+
+						out->type = PIPELINE_PORT_OUT_RING_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+					} else {
+						struct rte_port_ring_writer_nodrop_params
+							*params = &out->params.ring_nodrop;
+
+						out->type =
+							PIPELINE_PORT_OUT_RING_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz =
+							app->swq_params[in->id].burst_write;
+						params->n_retries =
+							app->swq_params[in->id].n_retries;
+					}
+				} else {
+					if (swq_params->dropless == 0) {
+						struct rte_port_ring_multi_writer_params *params =
+							&out->params.ring_multi;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+					} else {
+						struct rte_port_ring_multi_writer_nodrop_params
+							*params = &out->params.ring_multi_nodrop;
+
+						out->type = PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP;
+						params->ring = app->swq[in->id];
+						params->tx_burst_sz = swq_params->burst_write;
+						params->n_retries = swq_params->n_retries;
+					}
+				}
 			} else {
-				struct rte_port_ring_writer_nodrop_params
-					*params = &out->params.ring_nodrop;
-
-				out->type =
-					PIPELINE_PORT_OUT_RING_WRITER_NODROP;
-				params->ring = app->swq[in->id];
-				params->tx_burst_sz =
-					app->swq_params[in->id].burst_write;
-				params->n_retries =
-					app->swq_params[in->id].n_retries;
+				if (swq_params->ipv4_ras == 1) {
+					struct rte_port_ring_writer_ipv4_ras_params *params =
+						&out->params.ring_ipv4_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				} else {
+					struct rte_port_ring_writer_ipv6_ras_params *params =
+						&out->params.ring_ipv6_ras;
+
+					out->type = PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS;
+					params->ring = app->swq[in->id];
+					params->tx_burst_sz = swq_params->burst_write;
+				}
 			}
-			/* What about frag and ras ports? */
 			break;
+		}
 		case APP_PKTQ_OUT_TM: {
 			struct rte_port_sched_writer_params *params =
 				&out->params.sched;
diff --git a/examples/ip_pipeline/main.c b/examples/ip_pipeline/main.c
index 862e2f2..4944dcf 100644
--- a/examples/ip_pipeline/main.c
+++ b/examples/ip_pipeline/main.c
@@ -45,7 +45,9 @@ main(int argc, char **argv)
 
 	app_config_args(&app, argc, argv);
 
-	app_config_parse(&app, app.config_file);
+	app_config_preproc(&app);
+
+	app_config_parse(&app, app.parser_file);
 
 	app_config_check(&app);
 
diff --git a/examples/ip_pipeline/pipeline_be.h b/examples/ip_pipeline/pipeline_be.h
index 51f1e4f..f7269c0 100644
--- a/examples/ip_pipeline/pipeline_be.h
+++ b/examples/ip_pipeline/pipeline_be.h
@@ -45,6 +45,7 @@
 enum pipeline_port_in_type {
 	PIPELINE_PORT_IN_ETHDEV_READER,
 	PIPELINE_PORT_IN_RING_READER,
+	PIPELINE_PORT_IN_RING_MULTI_READER,
 	PIPELINE_PORT_IN_RING_READER_IPV4_FRAG,
 	PIPELINE_PORT_IN_RING_READER_IPV6_FRAG,
 	PIPELINE_PORT_IN_SCHED_READER,
@@ -56,6 +57,7 @@ struct pipeline_port_in_params {
 	union {
 		struct rte_port_ethdev_reader_params ethdev;
 		struct rte_port_ring_reader_params ring;
+		struct rte_port_ring_multi_reader_params ring_multi;
 		struct rte_port_ring_reader_ipv4_frag_params ring_ipv4_frag;
 		struct rte_port_ring_reader_ipv6_frag_params ring_ipv6_frag;
 		struct rte_port_sched_reader_params sched;
@@ -72,6 +74,8 @@ pipeline_port_in_params_convert(struct pipeline_port_in_params  *p)
 		return (void *) &p->params.ethdev;
 	case PIPELINE_PORT_IN_RING_READER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return (void *) &p->params.ring_ipv4_frag;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -93,6 +97,8 @@ pipeline_port_in_params_get_ops(struct pipeline_port_in_params  *p)
 		return &rte_port_ethdev_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER:
 		return &rte_port_ring_reader_ops;
+	case PIPELINE_PORT_IN_RING_MULTI_READER:
+		return &rte_port_ring_multi_reader_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV4_FRAG:
 		return &rte_port_ring_reader_ipv4_frag_ops;
 	case PIPELINE_PORT_IN_RING_READER_IPV6_FRAG:
@@ -110,7 +116,9 @@ enum pipeline_port_out_type {
 	PIPELINE_PORT_OUT_ETHDEV_WRITER,
 	PIPELINE_PORT_OUT_ETHDEV_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER,
 	PIPELINE_PORT_OUT_RING_WRITER_NODROP,
+	PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS,
 	PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS,
 	PIPELINE_PORT_OUT_SCHED_WRITER,
@@ -123,7 +131,9 @@ struct pipeline_port_out_params {
 		struct rte_port_ethdev_writer_params ethdev;
 		struct rte_port_ethdev_writer_nodrop_params ethdev_nodrop;
 		struct rte_port_ring_writer_params ring;
+		struct rte_port_ring_multi_writer_params ring_multi;
 		struct rte_port_ring_writer_nodrop_params ring_nodrop;
+		struct rte_port_ring_multi_writer_nodrop_params ring_multi_nodrop;
 		struct rte_port_ring_writer_ipv4_ras_params ring_ipv4_ras;
 		struct rte_port_ring_writer_ipv6_ras_params ring_ipv6_ras;
 		struct rte_port_sched_writer_params sched;
@@ -140,8 +150,12 @@ pipeline_port_out_params_convert(struct pipeline_port_out_params  *p)
 		return (void *) &p->params.ethdev_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return (void *) &p->params.ring;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return (void *) &p->params.ring_multi;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return (void *) &p->params.ring_nodrop;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return (void *) &p->params.ring_multi_nodrop;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return (void *) &p->params.ring_ipv4_ras;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
@@ -164,8 +178,12 @@ pipeline_port_out_params_get_ops(struct pipeline_port_out_params  *p)
 		return &rte_port_ethdev_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER:
 		return &rte_port_ring_writer_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER:
+		return &rte_port_ring_multi_writer_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_NODROP:
 		return &rte_port_ring_writer_nodrop_ops;
+	case PIPELINE_PORT_OUT_RING_MULTI_WRITER_NODROP:
+		return &rte_port_ring_multi_writer_nodrop_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV4_RAS:
 		return &rte_port_ring_writer_ipv4_ras_ops;
 	case PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS:
-- 
1.7.9.5

Re: [dpdk-dev] [PATCH v4 0/3] ip_pipeline: add MP/MC and frag/ras support to SWQs

[Thomas Monjalon]

2015-10-28 14:30, Piotr Azarewicz:
> This patch set enhancement ip_pipeline application:
> - librte_port: add support for multi-producer/multi-consumer ring ports
> - librte_port: bug fixes for ring ports with IPv4/IPv6 reassembly support
> - ip_pipeline application: integrate MP/MC and fragmentation/reassembly support to SWQs
> 
> v2 changes:
> - rte_port_ring:
> 	- fixed checkpatch errors
> 	- interlace the implementation of multi into the implementation of single
> 	- reduced the amount of code duplication
> 
> v3 changes:
> - new functions added in the .map
> - add a "Fixes:" tag in commit comment
> 
> v4 changes:
> - fix usage RTE_MBUF_METADATA_* macros
> 
> Acked-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
> 
> Piotr Azarewicz (3):
>   port: add mp/mc ring ports
>   port: fix ras/frag ring ports
>   examples/ip_pipeline: add mp/mc and frag/ras swq

Applied, thanks