[dpdk-dev] [PATCH v4] librte_pipeline: add support for packet redirection at action handlers

[Jasvinder Singh <jasvinder.singh@intel.com>]

Currently, there is no mechanism that allows the pipeline ports (in/out) and
table action handlers to override the default forwarding decision (as
previously configured per input port or in the table entry). Therefore, new
pipeline API functions have been added which allows action handlers to
hijack packets and remove them from the pipeline processing, and then either
drop them or send them out of the pipeline on any output port. The port
(in/out) and table action handler prototypes have been changed for making
use of these new API functions. This feature will be helpful to implement
functions such as exception handling (e.g. TTL =0), load balancing etc.
Changes are made to the ports and table action handlers defined in
app/test_pipeline and ip_pipeline sample application.

Signed-off-by: Jasvinder Singh <jasvinder.singh@intel.com>
Acked-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
v4
* merged library and app commits

v3
* improved comments in "rte_pipeline.h"

v2
* rebased on master

 app/test-pipeline/pipeline_acl.c                   |   3 +-
 app/test-pipeline/pipeline_hash.c                  |   3 +-
 app/test-pipeline/pipeline_lpm.c                   |   3 +-
 app/test-pipeline/pipeline_lpm_ipv6.c              |   3 +-
 app/test-pipeline/pipeline_stub.c                  |   3 +-
 doc/guides/rel_notes/deprecation.rst               |   5 -
 doc/guides/rel_notes/release_16_04.rst             |   6 +-
 .../ip_pipeline/pipeline/pipeline_actions_common.h |  47 ++-
 .../ip_pipeline/pipeline/pipeline_firewall_be.c    |   3 +-
 .../pipeline/pipeline_flow_actions_be.c            |   3 +-
 .../pipeline/pipeline_flow_classification_be.c     |   3 +-
 .../ip_pipeline/pipeline/pipeline_passthrough_be.c |   3 +-
 .../ip_pipeline/pipeline/pipeline_routing_be.c     |   3 +-
 lib/librte_pipeline/Makefile                       |   4 +-
 lib/librte_pipeline/rte_pipeline.c                 | 461 +++++++++++----------
 lib/librte_pipeline/rte_pipeline.h                 | 174 +++++---
 lib/librte_pipeline/rte_pipeline_version.map       |   8 +
 17 files changed, 399 insertions(+), 336 deletions(-)

diff --git a/app/test-pipeline/pipeline_acl.c b/app/test-pipeline/pipeline_acl.c
index f163e55..22d5f36 100644
--- a/app/test-pipeline/pipeline_acl.c
+++ b/app/test-pipeline/pipeline_acl.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -159,7 +159,6 @@ app_main_loop_worker_pipeline_acl(void) {
 			.ops = &rte_port_ring_writer_ops,
 			.arg_create = (void *) &port_ring_params,
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/app/test-pipeline/pipeline_hash.c b/app/test-pipeline/pipeline_hash.c
index 8b888d7..f8aac0d 100644
--- a/app/test-pipeline/pipeline_hash.c
+++ b/app/test-pipeline/pipeline_hash.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -140,7 +140,6 @@ app_main_loop_worker_pipeline_hash(void) {
 			.ops = &rte_port_ring_writer_ops,
 			.arg_create = (void *) &port_ring_params,
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/app/test-pipeline/pipeline_lpm.c b/app/test-pipeline/pipeline_lpm.c
index 2d7bc01..916abd4 100644
--- a/app/test-pipeline/pipeline_lpm.c
+++ b/app/test-pipeline/pipeline_lpm.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -99,7 +99,6 @@ app_main_loop_worker_pipeline_lpm(void) {
 			.ops = &rte_port_ring_writer_ops,
 			.arg_create = (void *) &port_ring_params,
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/app/test-pipeline/pipeline_lpm_ipv6.c b/app/test-pipeline/pipeline_lpm_ipv6.c
index c895b62..3352e89 100644
--- a/app/test-pipeline/pipeline_lpm_ipv6.c
+++ b/app/test-pipeline/pipeline_lpm_ipv6.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -100,7 +100,6 @@ app_main_loop_worker_pipeline_lpm_ipv6(void) {
 			.ops = &rte_port_ring_writer_ops,
 			.arg_create = (void *) &port_ring_params,
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/app/test-pipeline/pipeline_stub.c b/app/test-pipeline/pipeline_stub.c
index 0ad6f9b..ba710ca 100644
--- a/app/test-pipeline/pipeline_stub.c
+++ b/app/test-pipeline/pipeline_stub.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -94,7 +94,6 @@ app_main_loop_worker_pipeline_stub(void) {
 			.ops = &rte_port_ring_writer_ops,
 			.arg_create = (void *) &port_ring_params,
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/doc/guides/rel_notes/deprecation.rst b/doc/guides/rel_notes/deprecation.rst
index e94d4a2..1a7d660 100644
--- a/doc/guides/rel_notes/deprecation.rst
+++ b/doc/guides/rel_notes/deprecation.rst
@@ -40,11 +40,6 @@ Deprecation Notices
 * The scheduler statistics structure will change to allow keeping track of
   RED actions.
 
-* librte_pipeline: The prototype for the pipeline input port, output port
-  and table action handlers will be updated:
-  the pipeline parameter will be added, the packets mask parameter will be
-  either removed (for input port action handler) or made input-only.
-
 * ABI changes are planned in cmdline buffer size to allow the use of long
   commands (such as RETA update in testpmd).  This should impact
   CMDLINE_PARSE_RESULT_BUFSIZE, STR_TOKEN_SIZE and RDLINE_BUF_SIZE.
diff --git a/doc/guides/rel_notes/release_16_04.rst b/doc/guides/rel_notes/release_16_04.rst
index fd7dd1a..bd180ee 100644
--- a/doc/guides/rel_notes/release_16_04.rst
+++ b/doc/guides/rel_notes/release_16_04.rst
@@ -128,6 +128,10 @@ ABI Changes
   the previous releases and made in this release. Use fixed width quotes for
   ``rte_function_names`` or ``rte_struct_names``. Use the past tense.
 
+* librte_pipeline: The prototype for the pipeline input port, output port
+  and table action handlers are updated:the pipeline parameter is added,
+  the packets mask parameter has been either removed or made input-only.
+
 
 Shared Library Versions
 -----------------------
@@ -154,7 +158,7 @@ The libraries prepended with a plus sign were incremented in this version.
      librte_mbuf.so.2
      librte_mempool.so.1
      librte_meter.so.1
-     librte_pipeline.so.2
+   + librte_pipeline.so.3
      librte_pmd_bond.so.1
      librte_pmd_ring.so.2
      librte_port.so.2
diff --git a/examples/ip_pipeline/pipeline/pipeline_actions_common.h b/examples/ip_pipeline/pipeline/pipeline_actions_common.h
index aa1dd59..9958758 100644
--- a/examples/ip_pipeline/pipeline/pipeline_actions_common.h
+++ b/examples/ip_pipeline/pipeline/pipeline_actions_common.h
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2015 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -33,12 +33,19 @@
 #ifndef __INCLUDE_PIPELINE_ACTIONS_COMMON_H__
 #define __INCLUDE_PIPELINE_ACTIONS_COMMON_H__
 
+#include <stdint.h>
+
+#include <rte_common.h>
+#include <rte_cycles.h>
+#include <rte_mbuf.h>
+#include <rte_pipeline.h>
+
 #define PIPELINE_PORT_IN_AH(f_ah, f_pkt_work, f_pkt4_work)		\
 static int								\
 f_ah(									\
+	__rte_unused struct rte_pipeline *p,				\
 	struct rte_mbuf **pkts,						\
 	uint32_t n_pkts,						\
-	uint64_t *pkts_mask,						\
 	void *arg)							\
 {									\
 	uint32_t i;							\
@@ -49,21 +56,18 @@ f_ah(									\
 	for ( ; i < n_pkts; i++)					\
 		f_pkt_work(pkts[i], arg);				\
 									\
-	*pkts_mask = (~0LLU) >> (64 - n_pkts);				\
-									\
 	return 0;							\
 }
 
 #define PIPELINE_TABLE_AH_HIT(f_ah, f_pkt_work, f_pkt4_work)		\
 static int								\
 f_ah(									\
+	__rte_unused struct rte_pipeline *p,				\
 	struct rte_mbuf **pkts,						\
-	uint64_t *pkts_mask,						\
+	uint64_t pkts_in_mask,						\
 	struct rte_pipeline_table_entry **entries,			\
 	void *arg)							\
 {									\
-	uint64_t pkts_in_mask = *pkts_mask;				\
-									\
 	if ((pkts_in_mask & (pkts_in_mask + 1)) == 0) {			\
 		uint64_t n_pkts = __builtin_popcountll(pkts_in_mask);	\
 		uint32_t i;						\
@@ -88,13 +92,12 @@ f_ah(									\
 #define PIPELINE_TABLE_AH_MISS(f_ah, f_pkt_work, f_pkt4_work)		\
 static int								\
 f_ah(									\
+	__rte_unused struct rte_pipeline *p,				\
 	struct rte_mbuf **pkts,						\
-	uint64_t *pkts_mask,						\
+	uint64_t pkts_in_mask,						\
 	struct rte_pipeline_table_entry *entry,				\
 	void *arg)							\
 {									\
-	uint64_t pkts_in_mask = *pkts_mask;				\
-									\
 	if ((pkts_in_mask & (pkts_in_mask + 1)) == 0) {			\
 		uint64_t n_pkts = __builtin_popcountll(pkts_in_mask);	\
 		uint32_t i;						\
@@ -119,13 +122,14 @@ f_ah(									\
 #define PIPELINE_TABLE_AH_HIT_DROP_TIME(f_ah, f_pkt_work, f_pkt4_work)	\
 static int								\
 f_ah(									\
+	struct rte_pipeline *p,						\
 	struct rte_mbuf **pkts,						\
-	uint64_t *pkts_mask,						\
+	uint64_t pkts_mask,						\
 	struct rte_pipeline_table_entry **entries,			\
 	void *arg)							\
 {									\
-	uint64_t pkts_in_mask = *pkts_mask;				\
-	uint64_t pkts_out_mask = *pkts_mask;				\
+	uint64_t pkts_in_mask = pkts_mask;				\
+	uint64_t pkts_out_mask = pkts_mask;				\
 	uint64_t time = rte_rdtsc();					\
 									\
 	if ((pkts_in_mask & (pkts_in_mask + 1)) == 0) {			\
@@ -134,13 +138,13 @@ f_ah(									\
 									\
 		for (i = 0; i < (n_pkts & (~0x3LLU)); i += 4) {		\
 			uint64_t mask = f_pkt4_work(&pkts[i],		\
-				&entries[i], arg, time);	\
+				&entries[i], arg, time);		\
 			pkts_out_mask ^= mask << i;			\
 		}							\
 									\
 		for ( ; i < n_pkts; i++) {				\
 			uint64_t mask = f_pkt_work(pkts[i],		\
-				entries[i], arg, time);		\
+				entries[i], arg, time);			\
 			pkts_out_mask ^= mask << i;			\
 		}							\
 	} else								\
@@ -154,20 +158,22 @@ f_ah(									\
 			pkts_out_mask ^= mask << pos;			\
 		}							\
 									\
-	*pkts_mask = pkts_out_mask;					\
+	rte_pipeline_ah_packet_drop(p, pkts_out_mask ^ pkts_mask);	\
+									\
 	return 0;							\
 }
 
 #define PIPELINE_TABLE_AH_MISS_DROP_TIME(f_ah, f_pkt_work, f_pkt4_work)	\
 static int								\
 f_ah(									\
+	struct rte_pipeline *p,						\
 	struct rte_mbuf **pkts,						\
-	uint64_t *pkts_mask,						\
+	uint64_t pkts_mask,						\
 	struct rte_pipeline_table_entry *entry,				\
 	void *arg)							\
 {									\
-	uint64_t pkts_in_mask = *pkts_mask;				\
-	uint64_t pkts_out_mask = *pkts_mask;				\
+	uint64_t pkts_in_mask = pkts_mask;				\
+	uint64_t pkts_out_mask = pkts_mask;				\
 	uint64_t time = rte_rdtsc();					\
 									\
 	if ((pkts_in_mask & (pkts_in_mask + 1)) == 0) {			\
@@ -195,7 +201,8 @@ f_ah(									\
 			pkts_out_mask ^= mask << pos;			\
 		}							\
 									\
-	*pkts_mask = pkts_out_mask;					\
+	rte_pipeline_ah_packet_drop(p, pkts_out_mask ^ pkts_mask);	\
+									\
 	return 0;							\
 }
 
diff --git a/examples/ip_pipeline/pipeline/pipeline_firewall_be.c b/examples/ip_pipeline/pipeline/pipeline_firewall_be.c
index 1c376f7..0e8da94 100644
--- a/examples/ip_pipeline/pipeline/pipeline_firewall_be.c
+++ b/examples/ip_pipeline/pipeline/pipeline_firewall_be.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2015 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -439,7 +439,6 @@ pipeline_firewall_init(struct pipeline_params *params,
 			.arg_create = pipeline_port_out_params_convert(
 				&params->port_out[i]),
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/examples/ip_pipeline/pipeline/pipeline_flow_actions_be.c b/examples/ip_pipeline/pipeline/pipeline_flow_actions_be.c
index ec149c8..8244a5c 100644
--- a/examples/ip_pipeline/pipeline/pipeline_flow_actions_be.c
+++ b/examples/ip_pipeline/pipeline/pipeline_flow_actions_be.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2015 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -585,7 +585,6 @@ static void *pipeline_fa_init(struct pipeline_params *params,
 			.arg_create = pipeline_port_out_params_convert(
 				&params->port_out[i]),
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/examples/ip_pipeline/pipeline/pipeline_flow_classification_be.c b/examples/ip_pipeline/pipeline/pipeline_flow_classification_be.c
index ac80fc6..81b6454 100644
--- a/examples/ip_pipeline/pipeline/pipeline_flow_classification_be.c
+++ b/examples/ip_pipeline/pipeline/pipeline_flow_classification_be.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2015 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -414,7 +414,6 @@ static void *pipeline_fc_init(struct pipeline_params *params,
 			.arg_create = pipeline_port_out_params_convert(
 				&params->port_out[i]),
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/examples/ip_pipeline/pipeline/pipeline_passthrough_be.c b/examples/ip_pipeline/pipeline/pipeline_passthrough_be.c
index a898f7d..7642462 100644
--- a/examples/ip_pipeline/pipeline/pipeline_passthrough_be.c
+++ b/examples/ip_pipeline/pipeline/pipeline_passthrough_be.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2015 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -451,7 +451,6 @@ pipeline_passthrough_init(struct pipeline_params *params,
 			.arg_create = pipeline_port_out_params_convert(
 				&params->port_out[i]),
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/examples/ip_pipeline/pipeline/pipeline_routing_be.c b/examples/ip_pipeline/pipeline/pipeline_routing_be.c
index 9baabd0..a207286 100644
--- a/examples/ip_pipeline/pipeline/pipeline_routing_be.c
+++ b/examples/ip_pipeline/pipeline/pipeline_routing_be.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2015 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -1192,7 +1192,6 @@ pipeline_routing_init(struct pipeline_params *params,
 			.arg_create = pipeline_port_out_params_convert(
 				&params->port_out[i]),
 			.f_action = NULL,
-			.f_action_bulk = NULL,
 			.arg_ah = NULL,
 		};
 
diff --git a/lib/librte_pipeline/Makefile b/lib/librte_pipeline/Makefile
index 1166d3c..822fd41 100644
--- a/lib/librte_pipeline/Makefile
+++ b/lib/librte_pipeline/Makefile
@@ -1,6 +1,6 @@
 #   BSD LICENSE
 #
-#   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+#   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
 #   All rights reserved.
 #
 #   Redistribution and use in source and binary forms, with or without
@@ -41,7 +41,7 @@ CFLAGS += $(WERROR_FLAGS)
 
 EXPORT_MAP := rte_pipeline_version.map
 
-LIBABIVER := 2
+LIBABIVER := 3
 
 #
 # all source are stored in SRCS-y
diff --git a/lib/librte_pipeline/rte_pipeline.c b/lib/librte_pipeline/rte_pipeline.c
index d625fd2..7f8fbac 100644
--- a/lib/librte_pipeline/rte_pipeline.c
+++ b/lib/librte_pipeline/rte_pipeline.c
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -49,14 +49,30 @@
 #define RTE_TABLE_INVALID                                 UINT32_MAX
 
 #ifdef RTE_PIPELINE_STATS_COLLECT
-#define RTE_PIPELINE_STATS_ADD(counter, val) \
-	({ (counter) += (val); })
 
-#define RTE_PIPELINE_STATS_ADD_M(counter, mask) \
-	({ (counter) += __builtin_popcountll(mask); })
+#define RTE_PIPELINE_STATS_AH_DROP_WRITE(p, mask)			\
+	({ (p)->n_pkts_ah_drop = __builtin_popcountll(mask); })
+
+#define RTE_PIPELINE_STATS_AH_DROP_READ(p, counter)			\
+	({ (counter) += (p)->n_pkts_ah_drop; (p)->n_pkts_ah_drop = 0; })
+
+#define RTE_PIPELINE_STATS_TABLE_DROP0(p)				\
+	({ (p)->pkts_drop_mask = (p)->action_mask0[RTE_PIPELINE_ACTION_DROP]; })
+
+#define RTE_PIPELINE_STATS_TABLE_DROP1(p, counter)			\
+({									\
+	uint64_t mask = (p)->action_mask0[RTE_PIPELINE_ACTION_DROP];	\
+	mask ^= (p)->pkts_drop_mask;					\
+	(counter) += __builtin_popcountll(mask);			\
+})
+
 #else
-#define RTE_PIPELINE_STATS_ADD(counter, val)
-#define RTE_PIPELINE_STATS_ADD_M(counter, mask)
+
+#define RTE_PIPELINE_STATS_AH_DROP_WRITE(p, mask)
+#define RTE_PIPELINE_STATS_AH_DROP_READ(p, counter)
+#define RTE_PIPELINE_STATS_TABLE_DROP0(p)
+#define RTE_PIPELINE_STATS_TABLE_DROP1(p, counter)
+
 #endif
 
 struct rte_port_in {
@@ -75,6 +91,7 @@ struct rte_port_in {
 	/* List of enabled ports */
 	struct rte_port_in *next;
 
+	/* Statistics */
 	uint64_t n_pkts_dropped_by_ah;
 };
 
@@ -82,12 +99,12 @@ struct rte_port_out {
 	/* Input parameters */
 	struct rte_port_out_ops ops;
 	rte_pipeline_port_out_action_handler f_action;
-	rte_pipeline_port_out_action_handler_bulk f_action_bulk;
 	void *arg_ah;
 
 	/* Handle to low-level port */
 	void *h_port;
 
+	/* Statistics */
 	uint64_t n_pkts_dropped_by_ah;
 };
 
@@ -106,7 +123,7 @@ struct rte_table {
 	/* Handle to the low-level table object */
 	void *h_table;
 
-	/* Stats for this table. */
+	/* Statistics */
 	uint64_t n_pkts_dropped_by_lkp_hit_ah;
 	uint64_t n_pkts_dropped_by_lkp_miss_ah;
 	uint64_t n_pkts_dropped_lkp_hit;
@@ -133,13 +150,16 @@ struct rte_pipeline {
 
 	/* List of enabled ports */
 	uint64_t enabled_port_in_mask;
-	struct rte_port_in *port_in_first;
+	struct rte_port_in *port_in_next;
 
 	/* Pipeline run structures */
 	struct rte_mbuf *pkts[RTE_PORT_IN_BURST_SIZE_MAX];
 	struct rte_pipeline_table_entry *entries[RTE_PORT_IN_BURST_SIZE_MAX];
 	uint64_t action_mask0[RTE_PIPELINE_ACTIONS];
 	uint64_t action_mask1[RTE_PIPELINE_ACTIONS];
+	uint64_t pkts_mask;
+	uint64_t n_pkts_ah_drop;
+	uint64_t pkts_drop_mask;
 } __rte_cache_aligned;
 
 static inline uint32_t
@@ -234,7 +254,9 @@ rte_pipeline_create(struct rte_pipeline_params *params)
 	p->num_ports_out = 0;
 	p->num_tables = 0;
 	p->enabled_port_in_mask = 0;
-	p->port_in_first = NULL;
+	p->port_in_next = NULL;
+	p->pkts_mask = 0;
+	p->n_pkts_ah_drop = 0;
 
 	return p;
 }
@@ -759,9 +781,6 @@ rte_pipeline_port_out_check_params(struct rte_pipeline *p,
 		struct rte_pipeline_port_out_params *params,
 		uint32_t *port_id)
 {
-	rte_pipeline_port_out_action_handler f_ah;
-	rte_pipeline_port_out_action_handler_bulk f_ah_bulk;
-
 	if (p == NULL) {
 		RTE_LOG(ERR, PIPELINE, "%s: pipeline parameter NULL\n",
 			__func__);
@@ -794,7 +813,7 @@ rte_pipeline_port_out_check_params(struct rte_pipeline *p,
 
 	if (params->ops->f_tx == NULL) {
 		RTE_LOG(ERR, PIPELINE,
-				"%s: f_tx function pointer NULL\n", __func__);
+			"%s: f_tx function pointer NULL\n", __func__);
 		return -EINVAL;
 	}
 
@@ -804,15 +823,6 @@ rte_pipeline_port_out_check_params(struct rte_pipeline *p,
 		return -EINVAL;
 	}
 
-	f_ah = params->f_action;
-	f_ah_bulk = params->f_action_bulk;
-	if (((f_ah != NULL) && (f_ah_bulk == NULL)) ||
-	    ((f_ah == NULL) && (f_ah_bulk != NULL))) {
-		RTE_LOG(ERR, PIPELINE, "%s: Action handlers have to be either"
-			"both enabled or both disabled\n", __func__);
-		return -EINVAL;
-	}
-
 	/* Do we have room for one more port? */
 	if (p->num_ports_out == RTE_PIPELINE_PORT_OUT_MAX) {
 		RTE_LOG(ERR, PIPELINE,
@@ -905,7 +915,6 @@ rte_pipeline_port_out_create(struct rte_pipeline *p,
 	/* Save input parameters */
 	memcpy(&port->ops, params->ops, sizeof(struct rte_port_out_ops));
 	port->f_action = params->f_action;
-	port->f_action_bulk = params->f_action_bulk;
 	port->arg_ah = params->arg_ah;
 
 	/* Initialize port internal data structure */
@@ -959,9 +968,8 @@ int
 rte_pipeline_port_in_enable(struct rte_pipeline *p, uint32_t port_id)
 {
 	struct rte_port_in *port, *port_prev, *port_next;
-	struct rte_port_in *port_first, *port_last;
 	uint64_t port_mask;
-	uint32_t port_prev_id, port_next_id, port_first_id, port_last_id;
+	uint32_t port_prev_id, port_next_id;
 
 	/* Check input arguments */
 	if (p == NULL) {
@@ -977,6 +985,8 @@ rte_pipeline_port_in_enable(struct rte_pipeline *p, uint32_t port_id)
 		return -EINVAL;
 	}
 
+	port = &p->ports_in[port_id];
+
 	/* Return if current input port is already enabled */
 	port_mask = 1LLU << port_id;
 	if (p->enabled_port_in_mask & port_mask)
@@ -990,20 +1000,13 @@ rte_pipeline_port_in_enable(struct rte_pipeline *p, uint32_t port_id)
 
 	port_prev = &p->ports_in[port_prev_id];
 	port_next = &p->ports_in[port_next_id];
-	port = &p->ports_in[port_id];
 
 	port_prev->next = port;
 	port->next = port_next;
 
-	/* Update the first and last input ports in the chain */
-	port_first_id = __builtin_ctzll(p->enabled_port_in_mask);
-	port_last_id = 63 - __builtin_clzll(p->enabled_port_in_mask);
-
-	port_first = &p->ports_in[port_first_id];
-	port_last = &p->ports_in[port_last_id];
-
-	p->port_in_first = port_first;
-	port_last->next = NULL;
+	/* Check if list of enabled ports was previously empty */
+	if (p->enabled_port_in_mask == port_mask)
+		p->port_in_next = port;
 
 	return 0;
 }
@@ -1011,9 +1014,9 @@ rte_pipeline_port_in_enable(struct rte_pipeline *p, uint32_t port_id)
 int
 rte_pipeline_port_in_disable(struct rte_pipeline *p, uint32_t port_id)
 {
-	struct rte_port_in *port_prev, *port_next, *port_first, *port_last;
+	struct rte_port_in *port, *port_prev, *port_next;
 	uint64_t port_mask;
-	uint32_t port_prev_id, port_next_id, port_first_id, port_last_id;
+	uint32_t port_prev_id, port_next_id;
 
 	/* Check input arguments */
 	if (p == NULL) {
@@ -1028,15 +1031,18 @@ rte_pipeline_port_in_disable(struct rte_pipeline *p, uint32_t port_id)
 		return -EINVAL;
 	}
 
+	port = &p->ports_in[port_id];
+
 	/* Return if current input port is already disabled */
 	port_mask = 1LLU << port_id;
 	if ((p->enabled_port_in_mask & port_mask) == 0)
 		return 0;
 
+	p->enabled_port_in_mask &= ~port_mask;
+
 	/* Return if no other enabled ports */
-	if (__builtin_popcountll(p->enabled_port_in_mask) == 1) {
-		p->enabled_port_in_mask &= ~port_mask;
-		p->port_in_first = NULL;
+	if (p->enabled_port_in_mask == 0) {
+		p->port_in_next = NULL;
 
 		return 0;
 	}
@@ -1049,17 +1055,10 @@ rte_pipeline_port_in_disable(struct rte_pipeline *p, uint32_t port_id)
 	port_next = &p->ports_in[port_next_id];
 
 	port_prev->next = port_next;
-	p->enabled_port_in_mask &= ~port_mask;
-
-	/* Update the first and last input ports in the chain */
-	port_first_id = __builtin_ctzll(p->enabled_port_in_mask);
-	port_last_id = 63 - __builtin_clzll(p->enabled_port_in_mask);
-
-	port_first = &p->ports_in[port_first_id];
-	port_last = &p->ports_in[port_last_id];
 
-	p->port_in_first = port_first;
-	port_last->next = NULL;
+	/* Check if the port which has just been disabled is next to serve */
+	if (port == p->port_in_next)
+		p->port_in_next = port_next;
 
 	return 0;
 }
@@ -1149,28 +1148,32 @@ rte_pipeline_compute_masks(struct rte_pipeline *p, uint64_t pkts_mask)
 
 static inline void
 rte_pipeline_action_handler_port_bulk(struct rte_pipeline *p,
-		uint64_t pkts_mask, uint32_t port_id)
+	uint64_t pkts_mask, uint32_t port_id)
 {
 	struct rte_port_out *port_out = &p->ports_out[port_id];
 
+	p->pkts_mask = pkts_mask;
+
 	/* Output port user actions */
-	if (port_out->f_action_bulk != NULL) {
-		uint64_t mask = pkts_mask;
+	if (port_out->f_action != NULL) {
+		port_out->f_action(p, p->pkts, pkts_mask, port_out->arg_ah);
 
-		port_out->f_action_bulk(p->pkts, &pkts_mask, port_out->arg_ah);
-		p->action_mask0[RTE_PIPELINE_ACTION_DROP] |= pkts_mask ^  mask;
-		RTE_PIPELINE_STATS_ADD_M(port_out->n_pkts_dropped_by_ah,
-				pkts_mask ^  mask);
+		RTE_PIPELINE_STATS_AH_DROP_READ(p,
+			port_out->n_pkts_dropped_by_ah);
 	}
 
 	/* Output port TX */
-	if (pkts_mask != 0)
-		port_out->ops.f_tx_bulk(port_out->h_port, p->pkts, pkts_mask);
+	if (p->pkts_mask != 0)
+		port_out->ops.f_tx_bulk(port_out->h_port,
+			p->pkts,
+			p->pkts_mask);
 }
 
 static inline void
 rte_pipeline_action_handler_port(struct rte_pipeline *p, uint64_t pkts_mask)
 {
+	p->pkts_mask = pkts_mask;
+
 	if ((pkts_mask & (pkts_mask + 1)) == 0) {
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t i;
@@ -1185,18 +1188,18 @@ rte_pipeline_action_handler_port(struct rte_pipeline *p, uint64_t pkts_mask)
 			if (port_out->f_action == NULL) /* Output port TX */
 				port_out->ops.f_tx(port_out->h_port, pkt);
 			else {
-				uint64_t pkt_mask = 1LLU;
+				uint64_t pkt_mask = 1LLU << i;
 
-				port_out->f_action(pkt, &pkt_mask,
+				port_out->f_action(p,
+					p->pkts,
+					pkt_mask,
 					port_out->arg_ah);
-				p->action_mask0[RTE_PIPELINE_ACTION_DROP] |=
-					(pkt_mask ^ 1LLU) << i;
 
-				RTE_PIPELINE_STATS_ADD(port_out->n_pkts_dropped_by_ah,
-						pkt_mask ^ 1LLU);
+				RTE_PIPELINE_STATS_AH_DROP_READ(p,
+					port_out->n_pkts_dropped_by_ah);
 
 				/* Output port TX */
-				if (pkt_mask != 0)
+				if (pkt_mask & p->pkts_mask)
 					port_out->ops.f_tx(port_out->h_port,
 						pkt);
 			}
@@ -1221,18 +1224,16 @@ rte_pipeline_action_handler_port(struct rte_pipeline *p, uint64_t pkts_mask)
 			if (port_out->f_action == NULL) /* Output port TX */
 				port_out->ops.f_tx(port_out->h_port, pkt);
 			else {
-				pkt_mask = 1LLU;
-
-				port_out->f_action(pkt, &pkt_mask,
+				port_out->f_action(p,
+					p->pkts,
+					pkt_mask,
 					port_out->arg_ah);
-				p->action_mask0[RTE_PIPELINE_ACTION_DROP] |=
-					(pkt_mask ^ 1LLU) << i;
 
-				RTE_PIPELINE_STATS_ADD(port_out->n_pkts_dropped_by_ah,
-						pkt_mask ^ 1LLU);
+				RTE_PIPELINE_STATS_AH_DROP_READ(p,
+					port_out->n_pkts_dropped_by_ah);
 
 				/* Output port TX */
-				if (pkt_mask != 0)
+				if (pkt_mask & p->pkts_mask)
 					port_out->ops.f_tx(port_out->h_port,
 						pkt);
 			}
@@ -1244,6 +1245,8 @@ static inline void
 rte_pipeline_action_handler_port_meta(struct rte_pipeline *p,
 	uint64_t pkts_mask)
 {
+	p->pkts_mask = pkts_mask;
+
 	if ((pkts_mask & (pkts_mask + 1)) == 0) {
 		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
 		uint32_t i;
@@ -1260,18 +1263,18 @@ rte_pipeline_action_handler_port_meta(struct rte_pipeline *p,
 			if (port_out->f_action == NULL) /* Output port TX */
 				port_out->ops.f_tx(port_out->h_port, pkt);
 			else {
-				uint64_t pkt_mask = 1LLU;
+				uint64_t pkt_mask = 1LLU << i;
 
-				port_out->f_action(pkt, &pkt_mask,
+				port_out->f_action(p,
+					p->pkts,
+					pkt_mask,
 					port_out->arg_ah);
-				p->action_mask0[RTE_PIPELINE_ACTION_DROP] |=
-					(pkt_mask ^ 1LLU) << i;
 
-				RTE_PIPELINE_STATS_ADD(port_out->n_pkts_dropped_by_ah,
-						pkt_mask ^ 1ULL);
+				RTE_PIPELINE_STATS_AH_DROP_READ(p,
+					port_out->n_pkts_dropped_by_ah);
 
 				/* Output port TX */
-				if (pkt_mask != 0)
+				if (pkt_mask & p->pkts_mask)
 					port_out->ops.f_tx(port_out->h_port,
 						pkt);
 			}
@@ -1297,18 +1300,16 @@ rte_pipeline_action_handler_port_meta(struct rte_pipeline *p,
 			if (port_out->f_action == NULL) /* Output port TX */
 				port_out->ops.f_tx(port_out->h_port, pkt);
 			else {
-				pkt_mask = 1LLU;
-
-				port_out->f_action(pkt, &pkt_mask,
+				port_out->f_action(p,
+					p->pkts,
+					pkt_mask,
 					port_out->arg_ah);
-				p->action_mask0[RTE_PIPELINE_ACTION_DROP] |=
-					(pkt_mask ^ 1LLU) << i;
 
-				RTE_PIPELINE_STATS_ADD(port_out->n_pkts_dropped_by_ah,
-						pkt_mask ^ 1ULL);
+				RTE_PIPELINE_STATS_AH_DROP_READ(p,
+					port_out->n_pkts_dropped_by_ah);
 
 				/* Output port TX */
-				if (pkt_mask != 0)
+				if (pkt_mask & p->pkts_mask)
 					port_out->ops.f_tx(port_out->h_port,
 						pkt);
 			}
@@ -1342,136 +1343,140 @@ rte_pipeline_action_handler_drop(struct rte_pipeline *p, uint64_t pkts_mask)
 int
 rte_pipeline_run(struct rte_pipeline *p)
 {
-	struct rte_port_in *port_in;
-
-	for (port_in = p->port_in_first; port_in != NULL;
-		port_in = port_in->next) {
-		uint64_t pkts_mask;
-		uint32_t n_pkts, table_id;
-
-		/* Input port RX */
-		n_pkts = port_in->ops.f_rx(port_in->h_port, p->pkts,
-			port_in->burst_size);
-		if (n_pkts == 0)
-			continue;
-
-		pkts_mask = RTE_LEN2MASK(n_pkts, uint64_t);
-		p->action_mask0[RTE_PIPELINE_ACTION_DROP] = 0;
-		p->action_mask0[RTE_PIPELINE_ACTION_PORT] = 0;
-		p->action_mask0[RTE_PIPELINE_ACTION_PORT_META] = 0;
-		p->action_mask0[RTE_PIPELINE_ACTION_TABLE] = 0;
+	struct rte_port_in *port_in = p->port_in_next;
+	uint32_t n_pkts, table_id;
 
-		/* Input port user actions */
-		if (port_in->f_action != NULL) {
-			uint64_t mask = pkts_mask;
+	if (port_in == NULL)
+		return 0;
 
-			port_in->f_action(p->pkts, n_pkts, &pkts_mask, port_in->arg_ah);
-			mask ^= pkts_mask;
-			p->action_mask0[RTE_PIPELINE_ACTION_DROP] |= mask;
-			RTE_PIPELINE_STATS_ADD_M(port_in->n_pkts_dropped_by_ah, mask);
-		}
+	/* Input port RX */
+	n_pkts = port_in->ops.f_rx(port_in->h_port, p->pkts,
+		port_in->burst_size);
+	if (n_pkts == 0) {
+		p->port_in_next = port_in->next;
+		return 0;
+	}
 
-		/* Table */
-		for (table_id = port_in->table_id; pkts_mask != 0; ) {
-			struct rte_table *table;
-			uint64_t lookup_hit_mask, lookup_miss_mask;
-
-			/* Lookup */
-			table = &p->tables[table_id];
-			table->ops.f_lookup(table->h_table, p->pkts, pkts_mask,
-					&lookup_hit_mask, (void **) p->entries);
-			lookup_miss_mask = pkts_mask & (~lookup_hit_mask);
-
-			/* Lookup miss */
-			if (lookup_miss_mask != 0) {
-				struct rte_pipeline_table_entry *default_entry =
-					table->default_entry;
-
-				/* Table user actions */
-				if (table->f_action_miss != NULL) {
-					uint64_t mask = lookup_miss_mask;
-
-					table->f_action_miss(p->pkts,
-						&lookup_miss_mask,
-						default_entry, table->arg_ah);
-					mask ^= lookup_miss_mask;
-					p->action_mask0[RTE_PIPELINE_ACTION_DROP] |= mask;
-					RTE_PIPELINE_STATS_ADD_M(
-						table->n_pkts_dropped_by_lkp_miss_ah, mask);
-				}
-
-				/* Table reserved actions */
-				if ((default_entry->action ==
-					RTE_PIPELINE_ACTION_PORT) &&
-					(lookup_miss_mask != 0))
-					rte_pipeline_action_handler_port_bulk(p,
-						lookup_miss_mask,
-						default_entry->port_id);
-				else {
-					uint32_t pos = default_entry->action;
-
-					p->action_mask0[pos] = lookup_miss_mask;
-					if (pos == RTE_PIPELINE_ACTION_DROP) {
-						RTE_PIPELINE_STATS_ADD_M(table->n_pkts_dropped_lkp_miss,
-							lookup_miss_mask);
-					}
-				}
-			}
+	p->pkts_mask = RTE_LEN2MASK(n_pkts, uint64_t);
+	p->action_mask0[RTE_PIPELINE_ACTION_DROP] = 0;
+	p->action_mask0[RTE_PIPELINE_ACTION_PORT] = 0;
+	p->action_mask0[RTE_PIPELINE_ACTION_PORT_META] = 0;
+	p->action_mask0[RTE_PIPELINE_ACTION_TABLE] = 0;
+
+	/* Input port user actions */
+	if (port_in->f_action != NULL) {
+		port_in->f_action(p, p->pkts, n_pkts, port_in->arg_ah);
+
+		RTE_PIPELINE_STATS_AH_DROP_READ(p,
+			port_in->n_pkts_dropped_by_ah);
+	}
+
+	/* Table */
+	for (table_id = port_in->table_id; p->pkts_mask != 0; ) {
+		struct rte_table *table;
+		uint64_t lookup_hit_mask, lookup_miss_mask;
+
+		/* Lookup */
+		table = &p->tables[table_id];
+		table->ops.f_lookup(table->h_table, p->pkts, p->pkts_mask,
+			&lookup_hit_mask, (void **) p->entries);
+		lookup_miss_mask = p->pkts_mask & (~lookup_hit_mask);
+
+		/* Lookup miss */
+		if (lookup_miss_mask != 0) {
+			struct rte_pipeline_table_entry *default_entry =
+				table->default_entry;
+
+			p->pkts_mask = lookup_miss_mask;
 
-			/* Lookup hit */
-			if (lookup_hit_mask != 0) {
-				/* Table user actions */
-				if (table->f_action_hit != NULL) {
-					uint64_t mask = lookup_hit_mask;
-
-					table->f_action_hit(p->pkts,
-						&lookup_hit_mask,
-						p->entries, table->arg_ah);
-					mask ^= lookup_hit_mask;
-					p->action_mask0[RTE_PIPELINE_ACTION_DROP] |= mask;
-					RTE_PIPELINE_STATS_ADD_M(
-						table->n_pkts_dropped_by_lkp_hit_ah, mask);
-				}
-
-				/* Table reserved actions */
-				rte_pipeline_compute_masks(p, lookup_hit_mask);
-				p->action_mask0[RTE_PIPELINE_ACTION_DROP] |=
-					p->action_mask1[
-						RTE_PIPELINE_ACTION_DROP];
-				p->action_mask0[RTE_PIPELINE_ACTION_PORT] |=
-					p->action_mask1[
-						RTE_PIPELINE_ACTION_PORT];
-				p->action_mask0[RTE_PIPELINE_ACTION_PORT_META] |=
-					p->action_mask1[
-						RTE_PIPELINE_ACTION_PORT_META];
-				p->action_mask0[RTE_PIPELINE_ACTION_TABLE] |=
-					p->action_mask1[
-						RTE_PIPELINE_ACTION_TABLE];
-
-				RTE_PIPELINE_STATS_ADD_M(table->n_pkts_dropped_lkp_hit,
-						p->action_mask1[RTE_PIPELINE_ACTION_DROP]);
+			/* Table user actions */
+			if (table->f_action_miss != NULL) {
+				table->f_action_miss(p,
+					p->pkts,
+					lookup_miss_mask,
+					default_entry,
+					table->arg_ah);
+
+				RTE_PIPELINE_STATS_AH_DROP_READ(p,
+					table->n_pkts_dropped_by_lkp_miss_ah);
 			}
 
-			/* Prepare for next iteration */
-			pkts_mask = p->action_mask0[RTE_PIPELINE_ACTION_TABLE];
-			table_id = table->table_next_id;
-			p->action_mask0[RTE_PIPELINE_ACTION_TABLE] = 0;
+			/* Table reserved actions */
+			if ((default_entry->action == RTE_PIPELINE_ACTION_PORT) &&
+				(p->pkts_mask != 0))
+				rte_pipeline_action_handler_port_bulk(p,
+					p->pkts_mask,
+					default_entry->port_id);
+			else {
+				uint32_t pos = default_entry->action;
+
+				RTE_PIPELINE_STATS_TABLE_DROP0(p);
+
+				p->action_mask0[pos] |= p->pkts_mask;
+
+				RTE_PIPELINE_STATS_TABLE_DROP1(p,
+					table->n_pkts_dropped_lkp_miss);
+			}
 		}
 
-		/* Table reserved action PORT */
-		rte_pipeline_action_handler_port(p,
-				p->action_mask0[RTE_PIPELINE_ACTION_PORT]);
+		/* Lookup hit */
+		if (lookup_hit_mask != 0) {
+			p->pkts_mask = lookup_hit_mask;
 
-		/* Table reserved action PORT META */
-		rte_pipeline_action_handler_port_meta(p,
-				p->action_mask0[RTE_PIPELINE_ACTION_PORT_META]);
+			/* Table user actions */
+			if (table->f_action_hit != NULL) {
+				table->f_action_hit(p,
+					p->pkts,
+					lookup_hit_mask,
+					p->entries,
+					table->arg_ah);
 
-		/* Table reserved action DROP */
-		rte_pipeline_action_handler_drop(p,
-				p->action_mask0[RTE_PIPELINE_ACTION_DROP]);
+				RTE_PIPELINE_STATS_AH_DROP_READ(p,
+					table->n_pkts_dropped_by_lkp_hit_ah);
+			}
+
+			/* Table reserved actions */
+			RTE_PIPELINE_STATS_TABLE_DROP0(p);
+			rte_pipeline_compute_masks(p, p->pkts_mask);
+			p->action_mask0[RTE_PIPELINE_ACTION_DROP] |=
+				p->action_mask1[
+					RTE_PIPELINE_ACTION_DROP];
+			p->action_mask0[RTE_PIPELINE_ACTION_PORT] |=
+				p->action_mask1[
+					RTE_PIPELINE_ACTION_PORT];
+			p->action_mask0[RTE_PIPELINE_ACTION_PORT_META] |=
+				p->action_mask1[
+					RTE_PIPELINE_ACTION_PORT_META];
+			p->action_mask0[RTE_PIPELINE_ACTION_TABLE] |=
+				p->action_mask1[
+					RTE_PIPELINE_ACTION_TABLE];
+
+			RTE_PIPELINE_STATS_TABLE_DROP1(p,
+				table->n_pkts_dropped_lkp_hit);
+		}
+
+		/* Prepare for next iteration */
+		p->pkts_mask = p->action_mask0[RTE_PIPELINE_ACTION_TABLE];
+		table_id = table->table_next_id;
+		p->action_mask0[RTE_PIPELINE_ACTION_TABLE] = 0;
 	}
 
-	return 0;
+	/* Table reserved action PORT */
+	rte_pipeline_action_handler_port(p,
+		p->action_mask0[RTE_PIPELINE_ACTION_PORT]);
+
+	/* Table reserved action PORT META */
+	rte_pipeline_action_handler_port_meta(p,
+		p->action_mask0[RTE_PIPELINE_ACTION_PORT_META]);
+
+	/* Table reserved action DROP */
+	rte_pipeline_action_handler_drop(p,
+		p->action_mask0[RTE_PIPELINE_ACTION_DROP]);
+
+	/* Pick candidate for next port IN to serve */
+	p->port_in_next = port_in->next;
+
+	return (int) n_pkts;
 }
 
 int
@@ -1498,26 +1503,32 @@ rte_pipeline_flush(struct rte_pipeline *p)
 
 int
 rte_pipeline_port_out_packet_insert(struct rte_pipeline *p,
-		uint32_t port_id, struct rte_mbuf *pkt)
+	uint32_t port_id, struct rte_mbuf *pkt)
 {
 	struct rte_port_out *port_out = &p->ports_out[port_id];
 
-	/* Output port user actions */
-	if (port_out->f_action == NULL)
-		port_out->ops.f_tx(port_out->h_port, pkt); /* Output port TX */
-	else {
-		uint64_t pkt_mask = 1LLU;
-
-		port_out->f_action(pkt, &pkt_mask, port_out->arg_ah);
-
-		if (pkt_mask != 0) /* Output port TX */
-			port_out->ops.f_tx(port_out->h_port, pkt);
-		else {
-			rte_pktmbuf_free(pkt);
-			RTE_PIPELINE_STATS_ADD(port_out->n_pkts_dropped_by_ah, 1);
-		}
-	}
+	port_out->ops.f_tx(port_out->h_port, pkt); /* Output port TX */
+
+	return 0;
+}
+
+int rte_pipeline_ah_packet_hijack(struct rte_pipeline *p,
+	uint64_t pkts_mask)
+{
+	pkts_mask &= p->pkts_mask;
+	p->pkts_mask &= ~pkts_mask;
+
+	return 0;
+}
+
+int rte_pipeline_ah_packet_drop(struct rte_pipeline *p,
+	uint64_t pkts_mask)
+{
+	pkts_mask &= p->pkts_mask;
+	p->pkts_mask &= ~pkts_mask;
+	p->action_mask0[RTE_PIPELINE_ACTION_DROP] |= pkts_mask;
 
+	RTE_PIPELINE_STATS_AH_DROP_WRITE(p, pkts_mask);
 	return 0;
 }
 
diff --git a/lib/librte_pipeline/rte_pipeline.h b/lib/librte_pipeline/rte_pipeline.h
index 7302a62..84d1802 100644
--- a/lib/librte_pipeline/rte_pipeline.h
+++ b/lib/librte_pipeline/rte_pipeline.h
@@ -1,7 +1,7 @@
 /*-
  *   BSD LICENSE
  *
- *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
  *   All rights reserved.
  *
  *   Redistribution and use in source and binary forms, with or without
@@ -142,12 +142,12 @@ struct rte_pipeline_table_stats {
 	/** Number of packets dropped by lookup miss action handler. */
 	uint64_t n_pkts_dropped_by_lkp_miss_ah;
 
-	/** Number of packets dropped by pipeline in behalf of this table based on
-	 * on action specified in table entry. */
+	/** Number of packets dropped by pipeline in behalf of this
+	 * table based on action specified in table entry. */
 	uint64_t n_pkts_dropped_lkp_hit;
 
-	/** Number of packets dropped by pipeline in behalf of this table based on
-	 * on action specified in table entry. */
+	/** Number of packets dropped by pipeline in behalf of this
+	 *  table based on action specified in table entry. */
 	uint64_t n_pkts_dropped_lkp_miss;
 };
 
@@ -187,7 +187,7 @@ int rte_pipeline_check(struct rte_pipeline *p);
  * @param p
  *   Handle to pipeline instance
  * @return
- *   0 on success, error code otherwise
+ *   Number of packets read and processed
  */
 int rte_pipeline_run(struct rte_pipeline *p);
 
@@ -263,6 +263,8 @@ struct rte_pipeline_table_entry {
  * required not to free the packet buffer, which will be freed eventually by
  * the pipeline.
  *
+ * @param p
+ *   Handle to pipeline instance
  * @param pkts
  *   Burst of input packets specified as array of up to 64 pointers to struct
  *   rte_mbuf
@@ -283,8 +285,9 @@ struct rte_pipeline_table_entry {
  *   0 on success, error code otherwise
  */
 typedef int (*rte_pipeline_table_action_handler_hit)(
+	struct rte_pipeline *p,
 	struct rte_mbuf **pkts,
-	uint64_t *pkts_mask,
+	uint64_t pkts_mask,
 	struct rte_pipeline_table_entry **entries,
 	void *arg);
 
@@ -296,6 +299,8 @@ typedef int (*rte_pipeline_table_action_handler_hit)(
  * required not to free the packet buffer, which will be freed eventually by
  * the pipeline.
  *
+ * @param p
+ *   Handle to pipeline instance
  * @param pkts
  *   Burst of input packets specified as array of up to 64 pointers to struct
  *   rte_mbuf
@@ -316,8 +321,9 @@ typedef int (*rte_pipeline_table_action_handler_hit)(
  *   0 on success, error code otherwise
  */
 typedef int (*rte_pipeline_table_action_handler_miss)(
+	struct rte_pipeline *p,
 	struct rte_mbuf **pkts,
-	uint64_t *pkts_mask,
+	uint64_t pkts_mask,
 	struct rte_pipeline_table_entry *entry,
 	void *arg);
 
@@ -565,16 +571,14 @@ int rte_pipeline_table_stats_read(struct rte_pipeline *p, uint32_t table_id,
  * required not to free the packet buffer, which will be freed eventually by
  * the pipeline.
  *
+ * @param p
+ *   Handle to pipeline instance
  * @param pkts
  *   Burst of input packets specified as array of up to 64 pointers to struct
  *   rte_mbuf
  * @param n
  *   Number of packets in the input burst. This parameter specifies that
  *   elements 0 to (n-1) of pkts array are valid.
- * @param pkts_mask
- *   64-bit bitmask specifying which packets in the input burst are still valid
- *   after the action handler is executed. When pkts_mask bit n is set, then
- *   element n of pkts array is pointing to a valid packet.
  * @param arg
  *   Opaque parameter registered by the user at the pipeline table creation
  *   time
@@ -582,9 +586,9 @@ int rte_pipeline_table_stats_read(struct rte_pipeline *p, uint32_t table_id,
  *   0 on success, error code otherwise
  */
 typedef int (*rte_pipeline_port_in_action_handler)(
+	struct rte_pipeline *p,
 	struct rte_mbuf **pkts,
 	uint32_t n,
-	uint64_t *pkts_mask,
 	void *arg);
 
 /** Parameters for pipeline input port creation */
@@ -692,36 +696,15 @@ int rte_pipeline_port_in_stats_read(struct rte_pipeline *p, uint32_t port_id,
 #define RTE_PIPELINE_PORT_OUT_MAX                                   64
 
 /**
- * Pipeline output port action handler for single packet
- *
- * The action handler can decide to drop packets by resetting the pkt_mask
- * argument. In this case, the action handler is required not to free the
- * packet buffer, which will be freed eventually by the pipeline.
- *
- * @param pkt
- *   Input packet
- * @param pkt_mask
- *   Output argument set to 0 when the action handler decides to drop the input
- *   packet and to 1LLU otherwise
- * @param arg
- *   Opaque parameter registered by the user at the pipeline table creation
- *   time
- * @return
- *   0 on success, error code otherwise
- */
-typedef int (*rte_pipeline_port_out_action_handler)(
-	struct rte_mbuf *pkt,
-	uint64_t *pkt_mask,
-	void *arg);
-
-/**
- * Pipeline output port action handler bulk
+ * Pipeline output port action handler
  *
  * The action handler can decide to drop packets by resetting the associated
  * packet bit in the pkts_mask parameter. In this case, the action handler is
  * required not to free the packet buffer, which will be freed eventually by
  * the pipeline.
  *
+ * @param p
+ *   Handle to pipeline instance
  * @param pkts
  *   Burst of input packets specified as array of up to 64 pointers to struct
  *   rte_mbuf
@@ -735,9 +718,10 @@ typedef int (*rte_pipeline_port_out_action_handler)(
  * @return
  *   0 on success, error code otherwise
  */
-typedef int (*rte_pipeline_port_out_action_handler_bulk)(
+typedef int (*rte_pipeline_port_out_action_handler)(
+	struct rte_pipeline *p,
 	struct rte_mbuf **pkts,
-	uint64_t *pkts_mask,
+	uint64_t pkts_mask,
 	void *arg);
 
 /** Parameters for pipeline output port creation. The action handlers have to
@@ -750,12 +734,9 @@ struct rte_pipeline_port_out_params {
 	/** Opaque parameter to be passed to create operation when invoked */
 	void *arg_create;
 
-	/** Callback function executing the user actions on single input
-	packet */
-	rte_pipeline_port_out_action_handler f_action;
 	/** Callback function executing the user actions on bust of input
 	packets */
-	rte_pipeline_port_out_action_handler_bulk f_action_bulk;
+	rte_pipeline_port_out_action_handler f_action;
 	/** Opaque parameter to be passed to the action handler when invoked */
 	void *arg_ah;
 };
@@ -778,14 +759,38 @@ int rte_pipeline_port_out_create(struct rte_pipeline *p,
 	uint32_t *port_id);
 
 /**
- * Pipeline output port packet insert
+ * Read pipeline port out stats.
+ *
+ * This function reads port out statistics identified by *port_id* of given
+ * pipeline *p*.
+ *
+ * @param p
+ *   Handle to pipeline instance.
+ * @param port_id
+ *   Port ID what stats will be returned.
+ * @param stats
+ *   Statistics buffer.
+ * @param clear
+ *   If not 0 clear stats after reading.
+ * @return
+ *   0 on success, error code otherwise
+ */
+int rte_pipeline_port_out_stats_read(struct rte_pipeline *p, uint32_t port_id,
+	struct rte_pipeline_port_out_stats *stats, int clear);
+
+/*
+ * Functions to be called as part of the port IN/OUT or table action handlers
+ *
+ */
+/**
+ * Action handler packet insert to output port
  *
- * This function is called by the table action handler whenever it generates a
- * new packet to be sent out though one of the pipeline output ports. This
- * packet is not part of the burst of input packets read from any of the
- * pipeline input ports, so it is not an element of the pkts array input
- * parameter of the table action handler. This packet can be dropped by the
- * output port action handler.
+ * This function can be called by any input/output port or table action handler
+ * to send a packet out through one of the pipeline output ports. This packet is
+ * generated by the action handler, i.e. this packet is not part of the burst of
+ * packets read from one of the pipeline input ports and currently processed by
+ * the pipeline (this packet is not an element of the pkts array input parameter
+ * of the action handler).
  *
  * @param p
  *   Handle to pipeline instance
@@ -793,7 +798,7 @@ int rte_pipeline_port_out_create(struct rte_pipeline *p,
  *   Output port ID (returned by previous invocation of pipeline output port
  *   create) to send the packet specified by pkt
  * @param pkt
- *   New packet generated by the table action handler
+ *   New packet generated by the action handler
  * @return
  *   0 on success, error code otherwise
  */
@@ -801,25 +806,68 @@ int rte_pipeline_port_out_packet_insert(struct rte_pipeline *p,
 	uint32_t port_id,
 	struct rte_mbuf *pkt);
 
+#define rte_pipeline_ah_port_out_packet_insert \
+	rte_pipeline_port_out_packet_insert
+
 /**
- * Read pipeline port out stats.
+ * Action handler packet hijack
  *
- * This function reads port out statistics identified by *port_id* of given
- * pipeline *p*.
+ * This function can be called by any input/output port or table action handler
+ * to hijack selected packets from the burst of packets read from one of the
+ * pipeline input ports and currently processed by the pipeline. The hijacked
+ * packets are removed from any further pipeline processing, with the action
+ * handler now having the full ownership for these packets.
+ *
+ * The action handler can further send the hijacked packets out through any
+ * pipeline output port by calling the rte_pipeline_ah_port_out_packet_insert()
+ * function. The action handler can also drop these packets by calling the
+ * rte_pktmbuf_free() function, although a better alternative is provided by
+ * the action handler using the rte_pipeline_ah_packet_drop() function.
  *
  * @param p
- *   Handle to pipeline instance.
- * @param port_id
- *   Port ID what stats will be returned.
- * @param stats
- *   Statistics buffer.
- * @param clear
- *   If not 0 clear stats after reading.
+ *   Handle to pipeline instance
+ * @param pkts_mask
+ *   64-bit bitmask specifying which of the packets handed over for processing
+ *   to the action handler is to be hijacked by the action handler. When
+ *   pkts_mask bit n is set, then element n of the pkts array (input argument to
+ *   the action handler) is hijacked.
  * @return
  *   0 on success, error code otherwise
  */
-int rte_pipeline_port_out_stats_read(struct rte_pipeline *p, uint32_t port_id,
-	struct rte_pipeline_port_out_stats *stats, int clear);
+int rte_pipeline_ah_packet_hijack(struct rte_pipeline *p,
+	uint64_t pkts_mask);
+
+/**
+ * Action handler packet drop
+ *
+ * This function is called by the pipeline action handlers (port in/out, table)
+ * to drop the packets selected using packet mask.
+ *
+ * This function can be called by any input/output port or table action handler
+ * to drop selected packets from the burst of packets read from one of the
+ * pipeline input ports and currently processed by the pipeline. The dropped
+ * packets are removed from any further pipeline processing and the packet
+ * buffers are eventually freed to their buffer pool.
+ *
+ * This function updates the drop statistics counters correctly, therefore the
+ * recommended approach for dropping packets by the action handlers is to call
+ * this function as opposed to the action handler hijacking the packets first
+ * and then dropping them invisibly to the pipeline (by using the
+ * rte_pktmbuf_free() function).
+ *
+ * @param p
+ *   Handle to pipeline instance
+ * @param pkts_mask
+ *   64-bit bitmask specifying which of the packets handed over for processing
+ *   to the action handler is to be dropped by the action handler. When
+ *   pkts_mask bit n is set, then element n of the pkts array (input argument to
+ *   the action handler) is dropped.
+ * @return
+ *   0 on success, error code otherwise
+ */
+int rte_pipeline_ah_packet_drop(struct rte_pipeline *p,
+	uint64_t pkts_mask);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/librte_pipeline/rte_pipeline_version.map b/lib/librte_pipeline/rte_pipeline_version.map
index 4cc86f6..e4ee154 100644
--- a/lib/librte_pipeline/rte_pipeline_version.map
+++ b/lib/librte_pipeline/rte_pipeline_version.map
@@ -37,3 +37,11 @@ DPDK_2.2 {
 	rte_pipeline_table_entry_delete_bulk;
 
 } DPDK_2.1;
+
+DPDK_16.04 {
+	global:
+
+	rte_pipeline_ah_packet_hijack;
+	rte_pipeline_ah_packet_drop;
+
+} DPDK_2.2;
-- 
2.5.0