X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=examples%2Fip_pipeline%2Fthread.c;h=7fc03332eab53b54410ded910eebe6945fa6942d;hb=888904417d396a4b10ff673588b68b7e6ccccf64;hp=0291883d58275581b5a17fd4b6d45747a377fc8d;hpb=a3a95b7d58b9403d9d8bb6c87e2f42dbe212ac91;p=dpdk.git

diff --git a/examples/ip_pipeline/thread.c b/examples/ip_pipeline/thread.c
index 0291883d58..7fc03332ea 100644
--- a/examples/ip_pipeline/thread.c
+++ b/examples/ip_pipeline/thread.c
@@ -155,6 +155,29 @@ thread_init(void)
 	return 0;
 }
 
+static inline int
+thread_is_running(uint32_t thread_id)
+{
+	enum rte_lcore_state_t thread_state;
+
+	thread_state = rte_eal_get_lcore_state(thread_id);
+	return (thread_state == RUNNING) ? 1 : 0;
+}
+
+/**
+ * Pipeline is running when:
+ *    (A) Pipeline is mapped to a data plane thread AND
+ *    (B) Its data plane thread is in RUNNING state.
+ */
+static inline int
+pipeline_is_running(struct pipeline *p)
+{
+	if (p->enabled == 0)
+		return 0;
+
+	return thread_is_running(p->thread_id);
+}
+
 /**
  * Master thread & data plane threads: message passing
  */
@@ -254,6 +277,36 @@ thread_pipeline_enable(uint32_t thread_id,
 		p->enabled)
 		return -1;
 
+	if (!thread_is_running(thread_id)) {
+		struct thread_data *td = &thread_data[thread_id];
+		struct pipeline_data *tdp = &td->pipeline_data[td->n_pipelines];
+
+		if (td->n_pipelines >= THREAD_PIPELINES_MAX)
+			return -1;
+
+		/* Data plane thread */
+		td->p[td->n_pipelines] = p->p;
+
+		tdp->p = p->p;
+		for (i = 0; i < p->n_tables; i++)
+			tdp->table_data[i].a = p->table[i].a;
+
+		tdp->n_tables = p->n_tables;
+
+		tdp->msgq_req = p->msgq_req;
+		tdp->msgq_rsp = p->msgq_rsp;
+		tdp->timer_period = (rte_get_tsc_hz() * p->timer_period_ms) / 1000;
+		tdp->time_next = rte_get_tsc_cycles() + tdp->timer_period;
+
+		td->n_pipelines++;
+
+		/* Pipeline */
+		p->thread_id = thread_id;
+		p->enabled = 1;
+
+		return 0;
+	}
+
 	/* Allocate request */
 	req = thread_msg_alloc();
 	if (req == NULL)
@@ -316,6 +369,38 @@ thread_pipeline_disable(uint32_t thread_id,
 	if (p->thread_id != thread_id)
 		return -1;
 
+	if (!thread_is_running(thread_id)) {
+		struct thread_data *td = &thread_data[thread_id];
+		uint32_t i;
+
+		for (i = 0; i < td->n_pipelines; i++) {
+			struct pipeline_data *tdp = &td->pipeline_data[i];
+
+			if (tdp->p != p->p)
+				continue;
+
+			/* Data plane thread */
+			if (i < td->n_pipelines - 1) {
+				struct rte_pipeline *pipeline_last =
+					td->p[td->n_pipelines - 1];
+				struct pipeline_data *tdp_last =
+					&td->pipeline_data[td->n_pipelines - 1];
+
+				td->p[i] = pipeline_last;
+				memcpy(tdp, tdp_last, sizeof(*tdp));
+			}
+
+			td->n_pipelines--;
+
+			/* Pipeline */
+			p->enabled = 0;
+
+			break;
+		}
+
+		return 0;
+	}
+
 	/* Allocate request */
 	req = thread_msg_alloc();
 	if (req == NULL)
@@ -490,7 +575,15 @@ enum pipeline_req_type {
 	PIPELINE_REQ_TABLE_STATS_READ,
 	PIPELINE_REQ_TABLE_RULE_ADD,
 	PIPELINE_REQ_TABLE_RULE_ADD_DEFAULT,
-
+	PIPELINE_REQ_TABLE_RULE_ADD_BULK,
+	PIPELINE_REQ_TABLE_RULE_DELETE,
+	PIPELINE_REQ_TABLE_RULE_DELETE_DEFAULT,
+	PIPELINE_REQ_TABLE_RULE_STATS_READ,
+	PIPELINE_REQ_TABLE_MTR_PROFILE_ADD,
+	PIPELINE_REQ_TABLE_MTR_PROFILE_DELETE,
+	PIPELINE_REQ_TABLE_RULE_MTR_READ,
+	PIPELINE_REQ_TABLE_DSCP_TABLE_UPDATE,
+	PIPELINE_REQ_TABLE_RULE_TTL_READ,
 	PIPELINE_REQ_MAX
 };
 
@@ -515,6 +608,48 @@ struct pipeline_msg_req_table_rule_add_default {
 	struct table_rule_action action;
 };
 
+struct pipeline_msg_req_table_rule_add_bulk {
+	struct table_rule_match *match;
+	struct table_rule_action *action;
+	void **data;
+	uint32_t n_rules;
+	int bulk;
+};
+
+struct pipeline_msg_req_table_rule_delete {
+	struct table_rule_match match;
+};
+
+struct pipeline_msg_req_table_rule_stats_read {
+	void *data;
+	int clear;
+};
+
+struct pipeline_msg_req_table_mtr_profile_add {
+	uint32_t meter_profile_id;
+	struct rte_table_action_meter_profile profile;
+};
+
+struct pipeline_msg_req_table_mtr_profile_delete {
+	uint32_t meter_profile_id;
+};
+
+struct pipeline_msg_req_table_rule_mtr_read {
+	void *data;
+	uint32_t tc_mask;
+	int clear;
+};
+
+struct pipeline_msg_req_table_dscp_table_update {
+	uint64_t dscp_mask;
+	struct rte_table_action_dscp_table dscp_table;
+};
+
+struct pipeline_msg_req_table_rule_ttl_read {
+	void *data;
+	int clear;
+};
+
 struct pipeline_msg_req {
 	enum pipeline_req_type type;
 	uint32_t id; /* Port IN, port OUT or table ID */
@@ -526,6 +661,14 @@ struct pipeline_msg_req {
 		struct pipeline_msg_req_table_stats_read table_stats_read;
 		struct pipeline_msg_req_table_rule_add table_rule_add;
 		struct pipeline_msg_req_table_rule_add_default table_rule_add_default;
+		struct pipeline_msg_req_table_rule_add_bulk table_rule_add_bulk;
+		struct pipeline_msg_req_table_rule_delete table_rule_delete;
+		struct pipeline_msg_req_table_rule_stats_read table_rule_stats_read;
+		struct pipeline_msg_req_table_mtr_profile_add table_mtr_profile_add;
+		struct pipeline_msg_req_table_mtr_profile_delete table_mtr_profile_delete;
+		struct pipeline_msg_req_table_rule_mtr_read table_rule_mtr_read;
+		struct pipeline_msg_req_table_dscp_table_update table_dscp_table_update;
+		struct pipeline_msg_req_table_rule_ttl_read table_rule_ttl_read;
 	};
 };
 
@@ -549,6 +692,22 @@ struct pipeline_msg_rsp_table_rule_add_default {
 	void *data;
 };
 
+struct pipeline_msg_rsp_table_rule_add_bulk {
+	uint32_t n_rules;
+};
+
+struct pipeline_msg_rsp_table_rule_stats_read {
+	struct rte_table_action_stats_counters stats;
+};
+
+struct pipeline_msg_rsp_table_rule_mtr_read {
+	struct rte_table_action_mtr_counters stats;
+};
+
+struct pipeline_msg_rsp_table_rule_ttl_read {
+	struct rte_table_action_ttl_counters stats;
+};
+
 struct pipeline_msg_rsp {
 	int status;
 
@@ -559,6 +718,10 @@ struct pipeline_msg_rsp {
 		struct pipeline_msg_rsp_table_stats_read table_stats_read;
 		struct pipeline_msg_rsp_table_rule_add table_rule_add;
 		struct pipeline_msg_rsp_table_rule_add_default table_rule_add_default;
+		struct pipeline_msg_rsp_table_rule_add_bulk table_rule_add_bulk;
+		struct pipeline_msg_rsp_table_rule_stats_read table_rule_stats_read;
+		struct pipeline_msg_rsp_table_rule_mtr_read table_rule_mtr_read;
+		struct pipeline_msg_rsp_table_rule_ttl_read table_rule_ttl_read;
 	};
 };
 
@@ -620,10 +783,18 @@ pipeline_port_in_stats_read(const char *pipeline_name,
 
 	p = pipeline_find(pipeline_name);
 	if ((p == NULL) ||
-		(p->enabled == 0) ||
 		(port_id >= p->n_ports_in))
 		return -1;
 
+	if (!pipeline_is_running(p)) {
+		status = rte_pipeline_port_in_stats_read(p->p,
+			port_id,
+			stats,
+			clear);
+
+		return status;
+	}
+
 	/* Allocate request */
 	req = pipeline_msg_alloc();
 	if (req == NULL)
@@ -665,10 +836,14 @@ pipeline_port_in_enable(const char *pipeline_name,
 
 	p = pipeline_find(pipeline_name);
 	if ((p == NULL) ||
-		(p->enabled == 0) ||
 		(port_id >= p->n_ports_in))
 		return -1;
 
+	if (!pipeline_is_running(p)) {
+		status = rte_pipeline_port_in_enable(p->p, port_id);
+		return status;
+	}
+
 	/* Allocate request */
 	req = pipeline_msg_alloc();
 	if (req == NULL)
@@ -707,10 +882,14 @@ pipeline_port_in_disable(const char *pipeline_name,
 
 	p = pipeline_find(pipeline_name);
 	if ((p == NULL) ||
-		(p->enabled == 0) ||
 		(port_id >= p->n_ports_in))
 		return -1;
 
+	if (!pipeline_is_running(p)) {
+		status = rte_pipeline_port_in_disable(p->p, port_id);
+		return status;
+	}
+
 	/* Allocate request */
 	req = pipeline_msg_alloc();
 	if (req == NULL)
@@ -752,10 +931,18 @@ pipeline_port_out_stats_read(const char *pipeline_name,
 
 	p = pipeline_find(pipeline_name);
 	if ((p == NULL) ||
-		(p->enabled == 0) ||
 		(port_id >= p->n_ports_out))
 		return -1;
 
+	if (!pipeline_is_running(p)) {
+		status = rte_pipeline_port_out_stats_read(p->p,
+			port_id,
+			stats,
+			clear);
+
+		return status;
+	}
+
 	/* Allocate request */
 	req = pipeline_msg_alloc();
 	if (req == NULL)
@@ -800,10 +987,18 @@ pipeline_table_stats_read(const char *pipeline_name,
 
 	p = pipeline_find(pipeline_name);
 	if ((p == NULL) ||
-		(p->enabled == 0) ||
 		(table_id >= p->n_tables))
 		return -1;
 
+	if (!pipeline_is_running(p)) {
+		status = rte_pipeline_table_stats_read(p->p,
+			table_id,
+			stats,
+			clear);
+
+		return status;
+	}
+
 	/* Allocate request */
 	req = pipeline_msg_alloc();
 	if (req == NULL)
@@ -992,6 +1187,25 @@ action_default_check(struct table_rule_action *action,
 	return 0;
 }
 
+union table_rule_match_low_level {
+	struct rte_table_acl_rule_add_params acl_add;
+	struct rte_table_acl_rule_delete_params acl_delete;
+	struct rte_table_array_key array;
+	uint8_t hash[TABLE_RULE_MATCH_SIZE_MAX];
+	struct rte_table_lpm_key lpm_ipv4;
+	struct rte_table_lpm_ipv6_key lpm_ipv6;
+};
+
+static int
+match_convert(struct table_rule_match *mh,
+	union table_rule_match_low_level *ml,
+	int add);
+
+static int
+action_convert(struct rte_table_action *a,
+	struct table_rule_action *action,
+	struct rte_pipeline_table_entry *data);
+
 int
 pipeline_table_rule_add(const char *pipeline_name,
 	uint32_t table_id,
@@ -1013,12 +1227,56 @@ pipeline_table_rule_add(const char *pipeline_name,
 
 	p = pipeline_find(pipeline_name);
 	if ((p == NULL) ||
-		(p->enabled == 0) ||
 		(table_id >= p->n_tables) ||
 		match_check(match, p, table_id) ||
 		action_check(action, p, table_id))
 		return -1;
 
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+		union table_rule_match_low_level match_ll;
+		struct rte_pipeline_table_entry *data_in, *data_out;
+		int key_found;
+		uint8_t *buffer;
+
+		buffer = calloc(TABLE_RULE_ACTION_SIZE_MAX, sizeof(uint8_t));
+		if (buffer == NULL)
+			return -1;
+
+		/* Table match-action rule conversion */
+		data_in = (struct rte_pipeline_table_entry *)buffer;
+
+		status = match_convert(match, &match_ll, 1);
+		if (status) {
+			free(buffer);
+			return -1;
+		}
+
+		status = action_convert(a, action, data_in);
+		if (status) {
+			free(buffer);
+			return -1;
+		}
+
+		/* Add rule (match, action) to table */
+		status = rte_pipeline_table_entry_add(p->p,
+				table_id,
+				&match_ll,
+				data_in,
+				&key_found,
+				&data_out);
+		if (status) {
+			free(buffer);
+			return -1;
+		}
+
+		/* Write Response */
+		*data = data_out;
+
+		free(buffer);
+		return 0;
+	}
+
 	/* Allocate request */
 	req = pipeline_msg_alloc();
 	if (req == NULL)
@@ -1065,11 +1323,44 @@ pipeline_table_rule_add_default(const char *pipeline_name,
 
 	p = pipeline_find(pipeline_name);
 	if ((p == NULL) ||
-		(p->enabled == 0) ||
 		(table_id >= p->n_tables) ||
 		action_default_check(action, p, table_id))
 		return -1;
 
+	if (!pipeline_is_running(p)) {
+		struct rte_pipeline_table_entry *data_in, *data_out;
+		uint8_t *buffer;
+
+		buffer = calloc(TABLE_RULE_ACTION_SIZE_MAX, sizeof(uint8_t));
+		if (buffer == NULL)
+			return -1;
+
+		/* Apply actions */
+		data_in = (struct rte_pipeline_table_entry *)buffer;
+
+		data_in->action = action->fwd.action;
+		if (action->fwd.action == RTE_PIPELINE_ACTION_PORT)
+			data_in->port_id = action->fwd.id;
+		if (action->fwd.action == RTE_PIPELINE_ACTION_TABLE)
+			data_in->table_id = action->fwd.id;
+
+		/* Add default rule to table */
+		status = rte_pipeline_table_default_entry_add(p->p,
+				table_id,
+				data_in,
+				&data_out);
+		if (status) {
+			free(buffer);
+			return -1;
+		}
+
+		/* Write Response */
+		*data = data_out;
+
+		free(buffer);
+		return 0;
+	}
+
 	/* Allocate request */
 	req = pipeline_msg_alloc();
 	if (req == NULL)
@@ -1096,115 +1387,725 @@ pipeline_table_rule_add_default(const char *pipeline_name,
 	return status;
 }
 
-/**
- * Data plane threads: message handling
- */
-static inline struct pipeline_msg_req *
-pipeline_msg_recv(struct rte_ring *msgq_req)
+int
+pipeline_table_rule_add_bulk(const char *pipeline_name,
+	uint32_t table_id,
+	struct table_rule_match *match,
+	struct table_rule_action *action,
+	void **data,
+	uint32_t *n_rules)
 {
+	struct pipeline *p;
 	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	uint32_t i;
+	int status;
 
-	int status = rte_ring_sc_dequeue(msgq_req, (void **) &req);
+	/* Check input params */
+	if ((pipeline_name == NULL) ||
+		(match == NULL) ||
+		(action == NULL) ||
+		(data == NULL) ||
+		(n_rules == NULL) ||
+		(*n_rules == 0))
+		return -1;
 
-	if (status != 0)
-		return NULL;
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
 
-	return req;
-}
+	for (i = 0; i < *n_rules; i++)
+		if (match_check(match, p, table_id) ||
+			action_check(action, p, table_id))
+			return -1;
 
-static inline void
-pipeline_msg_send(struct rte_ring *msgq_rsp,
-	struct pipeline_msg_rsp *rsp)
-{
-	int status;
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+		union table_rule_match_low_level *match_ll;
+		uint8_t *action_ll;
+		void **match_ll_ptr;
+		struct rte_pipeline_table_entry **action_ll_ptr;
+		struct rte_pipeline_table_entry **entries_ptr =
+			(struct rte_pipeline_table_entry **)data;
+		uint32_t bulk =
+			(p->table[table_id].params.match_type == TABLE_ACL) ? 1 : 0;
+		int *found;
+
+		/* Memory allocation */
+		match_ll = calloc(*n_rules, sizeof(union table_rule_match_low_level));
+		action_ll = calloc(*n_rules, TABLE_RULE_ACTION_SIZE_MAX);
+		match_ll_ptr = calloc(*n_rules, sizeof(void *));
+		action_ll_ptr =
+			calloc(*n_rules, sizeof(struct rte_pipeline_table_entry *));
+		found = calloc(*n_rules, sizeof(int));
+
+		if (match_ll == NULL ||
+			action_ll == NULL ||
+			match_ll_ptr == NULL ||
+			action_ll_ptr == NULL ||
+			found == NULL)
+			goto fail;
+
+		for (i = 0; i < *n_rules; i++) {
+			match_ll_ptr[i] = (void *)&match_ll[i];
+			action_ll_ptr[i] =
+				(struct rte_pipeline_table_entry *)&action_ll[i * TABLE_RULE_ACTION_SIZE_MAX];
+		}
 
-	do {
-		status = rte_ring_sp_enqueue(msgq_rsp, rsp);
-	} while (status == -ENOBUFS);
-}
+		/* Rule match conversion */
+		for (i = 0; i < *n_rules; i++) {
+			status = match_convert(&match[i], match_ll_ptr[i], 1);
+			if (status)
+				goto fail;
+		}
 
-static struct pipeline_msg_rsp *
-pipeline_msg_handle_port_in_stats_read(struct pipeline_data *p,
-	struct pipeline_msg_req *req)
-{
-	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
-	uint32_t port_id = req->id;
-	int clear = req->port_in_stats_read.clear;
+		/* Rule action conversion */
+		for (i = 0; i < *n_rules; i++) {
+			status = action_convert(a, &action[i], action_ll_ptr[i]);
+			if (status)
+				goto fail;
+		}
 
-	rsp->status = rte_pipeline_port_in_stats_read(p->p,
-		port_id,
-		&rsp->port_in_stats_read.stats,
-		clear);
+		/* Add rule (match, action) to table */
+		if (bulk) {
+			status = rte_pipeline_table_entry_add_bulk(p->p,
+				table_id,
+				match_ll_ptr,
+				action_ll_ptr,
+				*n_rules,
+				found,
+				entries_ptr);
+			if (status)
+				*n_rules = 0;
+		} else {
+			for (i = 0; i < *n_rules; i++) {
+				status = rte_pipeline_table_entry_add(p->p,
+					table_id,
+					match_ll_ptr[i],
+					action_ll_ptr[i],
+					&found[i],
+					&entries_ptr[i]);
+				if (status) {
+					*n_rules = i;
+					break;
+				}
+			}
+		}
 
-	return rsp;
-}
+		/* Free */
+		free(found);
+		free(action_ll_ptr);
+		free(match_ll_ptr);
+		free(action_ll);
+		free(match_ll);
 
-static struct pipeline_msg_rsp *
-pipeline_msg_handle_port_in_enable(struct pipeline_data *p,
-	struct pipeline_msg_req *req)
-{
-	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
-	uint32_t port_id = req->id;
+		return status;
 
-	rsp->status = rte_pipeline_port_in_enable(p->p,
-		port_id);
+fail:
+		free(found);
+		free(action_ll_ptr);
+		free(match_ll_ptr);
+		free(action_ll);
+		free(match_ll);
 
-	return rsp;
-}
+		*n_rules = 0;
+		return -1;
+	}
 
-static struct pipeline_msg_rsp *
-pipeline_msg_handle_port_in_disable(struct pipeline_data *p,
-	struct pipeline_msg_req *req)
-{
-	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
-	uint32_t port_id = req->id;
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
 
-	rsp->status = rte_pipeline_port_in_disable(p->p,
-		port_id);
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_RULE_ADD_BULK;
+	req->id = table_id;
+	req->table_rule_add_bulk.match = match;
+	req->table_rule_add_bulk.action = action;
+	req->table_rule_add_bulk.data = data;
+	req->table_rule_add_bulk.n_rules = *n_rules;
+	req->table_rule_add_bulk.bulk =
+		(p->table[table_id].params.match_type == TABLE_ACL) ? 1 : 0;
 
-	return rsp;
-}
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
 
-static struct pipeline_msg_rsp *
-pipeline_msg_handle_port_out_stats_read(struct pipeline_data *p,
-	struct pipeline_msg_req *req)
-{
-	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
-	uint32_t port_id = req->id;
-	int clear = req->port_out_stats_read.clear;
+	/* Read response */
+	status = rsp->status;
+	if (status == 0)
+		*n_rules = rsp->table_rule_add_bulk.n_rules;
 
-	rsp->status = rte_pipeline_port_out_stats_read(p->p,
-		port_id,
-		&rsp->port_out_stats_read.stats,
-		clear);
+	/* Free response */
+	pipeline_msg_free(rsp);
 
-	return rsp;
+	return status;
 }
 
-static struct pipeline_msg_rsp *
-pipeline_msg_handle_table_stats_read(struct pipeline_data *p,
-	struct pipeline_msg_req *req)
+int
+pipeline_table_rule_delete(const char *pipeline_name,
+	uint32_t table_id,
+	struct table_rule_match *match)
 {
-	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
-	uint32_t port_id = req->id;
-	int clear = req->table_stats_read.clear;
-
-	rsp->status = rte_pipeline_table_stats_read(p->p,
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if ((pipeline_name == NULL) ||
+		(match == NULL))
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables) ||
+		match_check(match, p, table_id))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		union table_rule_match_low_level match_ll;
+		int key_found;
+
+		status = match_convert(match, &match_ll, 0);
+		if (status)
+			return -1;
+
+		status = rte_pipeline_table_entry_delete(p->p,
+				table_id,
+				&match_ll,
+				&key_found,
+				NULL);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_RULE_DELETE;
+	req->id = table_id;
+	memcpy(&req->table_rule_delete.match, match, sizeof(*match));
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+int
+pipeline_table_rule_delete_default(const char *pipeline_name,
+	uint32_t table_id)
+{
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if (pipeline_name == NULL)
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		status = rte_pipeline_table_default_entry_delete(p->p,
+			table_id,
+			NULL);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_RULE_DELETE_DEFAULT;
+	req->id = table_id;
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+int
+pipeline_table_rule_stats_read(const char *pipeline_name,
+	uint32_t table_id,
+	void *data,
+	struct rte_table_action_stats_counters *stats,
+	int clear)
+{
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if ((pipeline_name == NULL) ||
+		(data == NULL) ||
+		(stats == NULL))
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+
+		status = rte_table_action_stats_read(a,
+			data,
+			stats,
+			clear);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_RULE_STATS_READ;
+	req->id = table_id;
+	req->table_rule_stats_read.data = data;
+	req->table_rule_stats_read.clear = clear;
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+	if (status)
+		memcpy(stats, &rsp->table_rule_stats_read.stats, sizeof(*stats));
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+int
+pipeline_table_mtr_profile_add(const char *pipeline_name,
+	uint32_t table_id,
+	uint32_t meter_profile_id,
+	struct rte_table_action_meter_profile *profile)
+{
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if ((pipeline_name == NULL) ||
+		(profile == NULL))
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+
+		status = rte_table_action_meter_profile_add(a,
+			meter_profile_id,
+			profile);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_MTR_PROFILE_ADD;
+	req->id = table_id;
+	req->table_mtr_profile_add.meter_profile_id = meter_profile_id;
+	memcpy(&req->table_mtr_profile_add.profile, profile, sizeof(*profile));
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+int
+pipeline_table_mtr_profile_delete(const char *pipeline_name,
+	uint32_t table_id,
+	uint32_t meter_profile_id)
+{
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if (pipeline_name == NULL)
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+
+		status = rte_table_action_meter_profile_delete(a,
+				meter_profile_id);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_MTR_PROFILE_DELETE;
+	req->id = table_id;
+	req->table_mtr_profile_delete.meter_profile_id = meter_profile_id;
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+int
+pipeline_table_rule_mtr_read(const char *pipeline_name,
+	uint32_t table_id,
+	void *data,
+	uint32_t tc_mask,
+	struct rte_table_action_mtr_counters *stats,
+	int clear)
+{
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if ((pipeline_name == NULL) ||
+		(data == NULL) ||
+		(stats == NULL))
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+
+		status = rte_table_action_meter_read(a,
+				data,
+				tc_mask,
+				stats,
+				clear);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_RULE_MTR_READ;
+	req->id = table_id;
+	req->table_rule_mtr_read.data = data;
+	req->table_rule_mtr_read.tc_mask = tc_mask;
+	req->table_rule_mtr_read.clear = clear;
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+	if (status)
+		memcpy(stats, &rsp->table_rule_mtr_read.stats, sizeof(*stats));
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+int
+pipeline_table_dscp_table_update(const char *pipeline_name,
+	uint32_t table_id,
+	uint64_t dscp_mask,
+	struct rte_table_action_dscp_table *dscp_table)
+{
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if ((pipeline_name == NULL) ||
+		(dscp_table == NULL))
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+
+		status = rte_table_action_dscp_table_update(a,
+				dscp_mask,
+				dscp_table);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_DSCP_TABLE_UPDATE;
+	req->id = table_id;
+	req->table_dscp_table_update.dscp_mask = dscp_mask;
+	memcpy(&req->table_dscp_table_update.dscp_table,
+		dscp_table, sizeof(*dscp_table));
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+int
+pipeline_table_rule_ttl_read(const char *pipeline_name,
+	uint32_t table_id,
+	void *data,
+	struct rte_table_action_ttl_counters *stats,
+	int clear)
+{
+	struct pipeline *p;
+	struct pipeline_msg_req *req;
+	struct pipeline_msg_rsp *rsp;
+	int status;
+
+	/* Check input params */
+	if ((pipeline_name == NULL) ||
+		(data == NULL) ||
+		(stats == NULL))
+		return -1;
+
+	p = pipeline_find(pipeline_name);
+	if ((p == NULL) ||
+		(table_id >= p->n_tables))
+		return -1;
+
+	if (!pipeline_is_running(p)) {
+		struct rte_table_action *a = p->table[table_id].a;
+
+		status = rte_table_action_ttl_read(a,
+				data,
+				stats,
+				clear);
+
+		return status;
+	}
+
+	/* Allocate request */
+	req = pipeline_msg_alloc();
+	if (req == NULL)
+		return -1;
+
+	/* Write request */
+	req->type = PIPELINE_REQ_TABLE_RULE_TTL_READ;
+	req->id = table_id;
+	req->table_rule_ttl_read.data = data;
+	req->table_rule_ttl_read.clear = clear;
+
+	/* Send request and wait for response */
+	rsp = pipeline_msg_send_recv(p, req);
+	if (rsp == NULL)
+		return -1;
+
+	/* Read response */
+	status = rsp->status;
+	if (status)
+		memcpy(stats, &rsp->table_rule_ttl_read.stats, sizeof(*stats));
+
+	/* Free response */
+	pipeline_msg_free(rsp);
+
+	return status;
+}
+
+/**
+ * Data plane threads: message handling
+ */
+static inline struct pipeline_msg_req *
+pipeline_msg_recv(struct rte_ring *msgq_req)
+{
+	struct pipeline_msg_req *req;
+
+	int status = rte_ring_sc_dequeue(msgq_req, (void **) &req);
+
+	if (status != 0)
+		return NULL;
+
+	return req;
+}
+
+static inline void
+pipeline_msg_send(struct rte_ring *msgq_rsp,
+	struct pipeline_msg_rsp *rsp)
+{
+	int status;
+
+	do {
+		status = rte_ring_sp_enqueue(msgq_rsp, rsp);
+	} while (status == -ENOBUFS);
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_port_in_stats_read(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t port_id = req->id;
+	int clear = req->port_in_stats_read.clear;
+
+	rsp->status = rte_pipeline_port_in_stats_read(p->p,
+		port_id,
+		&rsp->port_in_stats_read.stats,
+		clear);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_port_in_enable(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t port_id = req->id;
+
+	rsp->status = rte_pipeline_port_in_enable(p->p,
+		port_id);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_port_in_disable(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t port_id = req->id;
+
+	rsp->status = rte_pipeline_port_in_disable(p->p,
+		port_id);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_port_out_stats_read(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t port_id = req->id;
+	int clear = req->port_out_stats_read.clear;
+
+	rsp->status = rte_pipeline_port_out_stats_read(p->p,
 		port_id,
-		&rsp->table_stats_read.stats,
+		&rsp->port_out_stats_read.stats,
 		clear);
 
 	return rsp;
 }
 
-union table_rule_match_low_level {
-	struct rte_table_acl_rule_add_params acl_add;
-	struct rte_table_acl_rule_delete_params acl_delete;
-	struct rte_table_array_key array;
-	uint8_t hash[TABLE_RULE_MATCH_SIZE_MAX];
-	struct rte_table_lpm_key lpm_ipv4;
-	struct rte_table_lpm_ipv6_key lpm_ipv6;
-};
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_stats_read(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t port_id = req->id;
+	int clear = req->table_stats_read.clear;
+
+	rsp->status = rte_pipeline_table_stats_read(p->p,
+		port_id,
+		&rsp->table_stats_read.stats,
+		clear);
+
+	return rsp;
+}
 
 static int
 match_convert_ipv6_depth(uint32_t depth, uint32_t *depth32)
@@ -1477,126 +2378,136 @@ match_convert(struct table_rule_match *mh,
 	}
 }
 
-static struct pipeline_msg_rsp *
-pipeline_msg_handle_table_rule_add(struct pipeline_data *p,
-	struct pipeline_msg_req *req)
+static int
+action_convert(struct rte_table_action *a,
+	struct table_rule_action *action,
+	struct rte_pipeline_table_entry *data)
 {
-	union table_rule_match_low_level match_ll;
-	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
-	struct table_rule_match *match = &req->table_rule_add.match;
-	struct table_rule_action *action = &req->table_rule_add.action;
-	struct rte_pipeline_table_entry *data_in, *data_out;
-	uint32_t table_id = req->id;
-	int key_found, status;
-	struct rte_table_action *a = p->table_data[table_id].a;
+	int status;
 
 	/* Apply actions */
-	memset(p->buffer, 0, sizeof(p->buffer));
-	data_in = (struct rte_pipeline_table_entry *) p->buffer;
-
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_FWD,
 			&action->fwd);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
+	}
+
+	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_LB)) {
+		status = rte_table_action_apply(a,
+			data,
+			RTE_TABLE_ACTION_LB,
+			&action->lb);
+
+		if (status)
+			return status;
 	}
 
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_MTR)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_MTR,
 			&action->mtr);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
 	}
 
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TM)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_TM,
 			&action->tm);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
 	}
 
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_ENCAP)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_ENCAP,
 			&action->encap);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
 	}
 
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_NAT)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_NAT,
 			&action->nat);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
 	}
 
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TTL)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_TTL,
 			&action->ttl);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
 	}
 
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_STATS)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_STATS,
 			&action->stats);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
 	}
 
 	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TIME)) {
 		status = rte_table_action_apply(a,
-			data_in,
+			data,
 			RTE_TABLE_ACTION_TIME,
 			&action->time);
 
-		if (status) {
-			rsp->status = -1;
-			return rsp;
-		}
+		if (status)
+			return status;
 	}
 
-	/* Add rule (match, action) to table */
+	return 0;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_rule_add(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	union table_rule_match_low_level match_ll;
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	struct table_rule_match *match = &req->table_rule_add.match;
+	struct table_rule_action *action = &req->table_rule_add.action;
+	struct rte_pipeline_table_entry *data_in, *data_out;
+	uint32_t table_id = req->id;
+	int key_found, status;
+	struct rte_table_action *a = p->table_data[table_id].a;
+
+	/* Apply actions */
+	memset(p->buffer, 0, sizeof(p->buffer));
+	data_in = (struct rte_pipeline_table_entry *) p->buffer;
+
 	status = match_convert(match, &match_ll, 1);
 	if (status) {
 		rsp->status = -1;
 		return rsp;
 	}
 
+	status = action_convert(a, action, data_in);
+	if (status) {
+		rsp->status = -1;
+		return rsp;
+	}
+
 	status = rte_pipeline_table_entry_add(p->p,
 		table_id,
 		&match_ll,
@@ -1652,6 +2563,261 @@ pipeline_msg_handle_table_rule_add_default(struct pipeline_data *p,
 	return rsp;
 }
 
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_rule_add_bulk(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+
+	uint32_t table_id = req->id;
+	struct table_rule_match *match = req->table_rule_add_bulk.match;
+	struct table_rule_action *action = req->table_rule_add_bulk.action;
+	struct rte_pipeline_table_entry **data =
+		(struct rte_pipeline_table_entry **)req->table_rule_add_bulk.data;
+	uint32_t n_rules = req->table_rule_add_bulk.n_rules;
+	uint32_t bulk = req->table_rule_add_bulk.bulk;
+
+	struct rte_table_action *a = p->table_data[table_id].a;
+	union table_rule_match_low_level *match_ll;
+	uint8_t *action_ll;
+	void **match_ll_ptr;
+	struct rte_pipeline_table_entry **action_ll_ptr;
+	int *found, status;
+	uint32_t i;
+
+	/* Memory allocation */
+	match_ll = calloc(n_rules, sizeof(union table_rule_match_low_level));
+	action_ll = calloc(n_rules, TABLE_RULE_ACTION_SIZE_MAX);
+	match_ll_ptr = calloc(n_rules, sizeof(void *));
+	action_ll_ptr =
+		calloc(n_rules, sizeof(struct rte_pipeline_table_entry *));
+	found = calloc(n_rules, sizeof(int));
+
+	if ((match_ll == NULL) ||
+		(action_ll == NULL) ||
+		(match_ll_ptr == NULL) ||
+		(action_ll_ptr == NULL) ||
+		(found == NULL))
+		goto fail;
+
+	for (i = 0; i < n_rules; i++) {
+		match_ll_ptr[i] = (void *)&match_ll[i];
+		action_ll_ptr[i] =
+			(struct rte_pipeline_table_entry *)&action_ll[i * TABLE_RULE_ACTION_SIZE_MAX];
+	}
+
+	/* Rule match conversion */
+	for (i = 0; i < n_rules; i++) {
+		status = match_convert(&match[i], match_ll_ptr[i], 1);
+		if (status)
+			goto fail;
+	}
+
+	/* Rule action conversion */
+	for (i = 0; i < n_rules; i++) {
+		status = action_convert(a, &action[i], action_ll_ptr[i]);
+		if (status)
+			goto fail;
+	}
+
+	/* Add rule (match, action) to table */
+	if (bulk) {
+		status = rte_pipeline_table_entry_add_bulk(p->p,
+			table_id,
+			match_ll_ptr,
+			action_ll_ptr,
+			n_rules,
+			found,
+			data);
+		if (status)
+			n_rules = 0;
+	} else
+		for (i = 0; i < n_rules; i++) {
+			status = rte_pipeline_table_entry_add(p->p,
+				table_id,
+				match_ll_ptr[i],
+				action_ll_ptr[i],
+				&found[i],
+				&data[i]);
+			if (status) {
+				n_rules = i;
+				break;
+			}
+		}
+
+	/* Write response */
+	rsp->status = 0;
+	rsp->table_rule_add_bulk.n_rules = n_rules;
+
+	/* Free */
+	free(found);
+	free(action_ll_ptr);
+	free(match_ll_ptr);
+	free(action_ll);
+	free(match_ll);
+
+	return rsp;
+
+fail:
+	free(found);
+	free(action_ll_ptr);
+	free(match_ll_ptr);
+	free(action_ll);
+	free(match_ll);
+
+	rsp->status = -1;
+	rsp->table_rule_add_bulk.n_rules = 0;
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_rule_delete(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	union table_rule_match_low_level match_ll;
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	struct table_rule_match *match = &req->table_rule_delete.match;
+	uint32_t table_id = req->id;
+	int key_found, status;
+
+	status = match_convert(match, &match_ll, 0);
+	if (status) {
+		rsp->status = -1;
+		return rsp;
+	}
+
+	rsp->status = rte_pipeline_table_entry_delete(p->p,
+		table_id,
+		&match_ll,
+		&key_found,
+		NULL);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_rule_delete_default(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t table_id = req->id;
+
+	rsp->status = rte_pipeline_table_default_entry_delete(p->p,
+		table_id,
+		NULL);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_rule_stats_read(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t table_id = req->id;
+	void *data = req->table_rule_stats_read.data;
+	int clear = req->table_rule_stats_read.clear;
+	struct rte_table_action *a = p->table_data[table_id].a;
+
+	rsp->status = rte_table_action_stats_read(a,
+		data,
+		&rsp->table_rule_stats_read.stats,
+		clear);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_mtr_profile_add(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t table_id = req->id;
+	uint32_t meter_profile_id = req->table_mtr_profile_add.meter_profile_id;
+	struct rte_table_action_meter_profile *profile =
+		&req->table_mtr_profile_add.profile;
+	struct rte_table_action *a = p->table_data[table_id].a;
+
+	rsp->status = rte_table_action_meter_profile_add(a,
+		meter_profile_id,
+		profile);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_mtr_profile_delete(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t table_id = req->id;
+	uint32_t meter_profile_id =
+		req->table_mtr_profile_delete.meter_profile_id;
+	struct rte_table_action *a = p->table_data[table_id].a;
+
+	rsp->status = rte_table_action_meter_profile_delete(a,
+		meter_profile_id);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_rule_mtr_read(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t table_id = req->id;
+	void *data = req->table_rule_mtr_read.data;
+	uint32_t tc_mask = req->table_rule_mtr_read.tc_mask;
+	int clear = req->table_rule_mtr_read.clear;
+	struct rte_table_action *a = p->table_data[table_id].a;
+
+	rsp->status = rte_table_action_meter_read(a,
+		data,
+		tc_mask,
+		&rsp->table_rule_mtr_read.stats,
+		clear);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_dscp_table_update(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t table_id = req->id;
+	uint64_t dscp_mask = req->table_dscp_table_update.dscp_mask;
+	struct rte_table_action_dscp_table *dscp_table =
+		&req->table_dscp_table_update.dscp_table;
+	struct rte_table_action *a = p->table_data[table_id].a;
+
+	rsp->status = rte_table_action_dscp_table_update(a,
+		dscp_mask,
+		dscp_table);
+
+	return rsp;
+}
+
+static struct pipeline_msg_rsp *
+pipeline_msg_handle_table_rule_ttl_read(struct pipeline_data *p,
+	struct pipeline_msg_req *req)
+{
+	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
+	uint32_t table_id = req->id;
+	void *data = req->table_rule_ttl_read.data;
+	int clear = req->table_rule_ttl_read.clear;
+	struct rte_table_action *a = p->table_data[table_id].a;
+
+	rsp->status = rte_table_action_ttl_read(a,
+		data,
+		&rsp->table_rule_ttl_read.stats,
+		clear);
+
+	return rsp;
+}
+
 static void
 pipeline_msg_handle(struct pipeline_data *p)
 {
@@ -1692,6 +2858,42 @@ pipeline_msg_handle(struct pipeline_data *p)
 			rsp = pipeline_msg_handle_table_rule_add_default(p,	req);
 			break;
 
+		case PIPELINE_REQ_TABLE_RULE_ADD_BULK:
+			rsp = pipeline_msg_handle_table_rule_add_bulk(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_RULE_DELETE:
+			rsp = pipeline_msg_handle_table_rule_delete(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_RULE_DELETE_DEFAULT:
+			rsp = pipeline_msg_handle_table_rule_delete_default(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_RULE_STATS_READ:
+			rsp = pipeline_msg_handle_table_rule_stats_read(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_MTR_PROFILE_ADD:
+			rsp = pipeline_msg_handle_table_mtr_profile_add(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_MTR_PROFILE_DELETE:
+			rsp = pipeline_msg_handle_table_mtr_profile_delete(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_RULE_MTR_READ:
+			rsp = pipeline_msg_handle_table_rule_mtr_read(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_DSCP_TABLE_UPDATE:
+			rsp = pipeline_msg_handle_table_dscp_table_update(p, req);
+			break;
+
+		case PIPELINE_REQ_TABLE_RULE_TTL_READ:
+			rsp = pipeline_msg_handle_table_rule_ttl_read(p, req);
+			break;
+
 		default:
 			rsp = (struct pipeline_msg_rsp *) req;
 			rsp->status = -1;