numam-dpdk/examples/ip_pipeline/thread.c

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright(c) 2010-2018 Intel Corporation
 */

#include <stdlib.h>

#include <rte_common.h>
#include <rte_cycles.h>
#include <rte_lcore.h>
#include <rte_ring.h>

#include <rte_table_acl.h>
#include <rte_table_array.h>
#include <rte_table_hash.h>
#include <rte_table_lpm.h>
#include <rte_table_lpm_ipv6.h>

#include "common.h"
#include "thread.h"
#include "pipeline.h"

#ifndef THREAD_PIPELINES_MAX
#define THREAD_PIPELINES_MAX                               256
#endif

#ifndef THREAD_MSGQ_SIZE
#define THREAD_MSGQ_SIZE                                   64
#endif

#ifndef THREAD_TIMER_PERIOD_MS
#define THREAD_TIMER_PERIOD_MS                             100
#endif

/**
 * Main thread: data plane thread context
 */
struct thread {
	struct rte_ring *msgq_req;
	struct rte_ring *msgq_rsp;

	uint32_t enabled;
};

static struct thread thread[RTE_MAX_LCORE];

/**
 * Data plane threads: context
 */
struct table_data {
	struct rte_table_action *a;
};

struct pipeline_data {
	struct rte_pipeline *p;
	struct table_data table_data[RTE_PIPELINE_TABLE_MAX];
	uint32_t n_tables;

	struct rte_ring *msgq_req;
	struct rte_ring *msgq_rsp;
	uint64_t timer_period; /* Measured in CPU cycles. */
	uint64_t time_next;

	uint8_t buffer[TABLE_RULE_ACTION_SIZE_MAX];
};

struct thread_data {
	struct rte_pipeline *p[THREAD_PIPELINES_MAX];
	uint32_t n_pipelines;

	struct pipeline_data pipeline_data[THREAD_PIPELINES_MAX];
	struct rte_ring *msgq_req;
	struct rte_ring *msgq_rsp;
	uint64_t timer_period; /* Measured in CPU cycles. */
	uint64_t time_next;
	uint64_t time_next_min;
} __rte_cache_aligned;

static struct thread_data thread_data[RTE_MAX_LCORE];

/**
 * Main thread: data plane thread init
 */
static void
thread_free(void)
{
	uint32_t i;

	for (i = 0; i < RTE_MAX_LCORE; i++) {
		struct thread *t = &thread[i];

		if (!rte_lcore_is_enabled(i))
			continue;

		/* MSGQs */
		if (t->msgq_req)
			rte_ring_free(t->msgq_req);

		if (t->msgq_rsp)
			rte_ring_free(t->msgq_rsp);
	}
}

int
thread_init(void)
{
	uint32_t i;

	RTE_LCORE_FOREACH_WORKER(i) {
		char name[NAME_MAX];
		struct rte_ring *msgq_req, *msgq_rsp;
		struct thread *t = &thread[i];
		struct thread_data *t_data = &thread_data[i];
		uint32_t cpu_id = rte_lcore_to_socket_id(i);

		/* MSGQs */
		snprintf(name, sizeof(name), "THREAD-%04x-MSGQ-REQ", i);

		msgq_req = rte_ring_create(name,
			THREAD_MSGQ_SIZE,
			cpu_id,
			RING_F_SP_ENQ | RING_F_SC_DEQ);

		if (msgq_req == NULL) {
			thread_free();
			return -1;
		}

		snprintf(name, sizeof(name), "THREAD-%04x-MSGQ-RSP", i);

		msgq_rsp = rte_ring_create(name,
			THREAD_MSGQ_SIZE,
			cpu_id,
			RING_F_SP_ENQ | RING_F_SC_DEQ);

		if (msgq_rsp == NULL) {
			thread_free();
			return -1;
		}

		/* Main thread records */
		t->msgq_req = msgq_req;
		t->msgq_rsp = msgq_rsp;
		t->enabled = 1;

		/* Data plane thread records */
		t_data->n_pipelines = 0;
		t_data->msgq_req = msgq_req;
		t_data->msgq_rsp = msgq_rsp;
		t_data->timer_period =
			(rte_get_tsc_hz() * THREAD_TIMER_PERIOD_MS) / 1000;
		t_data->time_next = rte_get_tsc_cycles() + t_data->timer_period;
		t_data->time_next_min = t_data->time_next;
	}

	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);
}

/**
 * Main thread & data plane threads: message passing
 */
enum thread_req_type {
	THREAD_REQ_PIPELINE_ENABLE = 0,
	THREAD_REQ_PIPELINE_DISABLE,
	THREAD_REQ_MAX
};

struct thread_msg_req {
	enum thread_req_type type;

	union {
		struct {
			struct rte_pipeline *p;
			struct {
				struct rte_table_action *a;
			} table[RTE_PIPELINE_TABLE_MAX];
			struct rte_ring *msgq_req;
			struct rte_ring *msgq_rsp;
			uint32_t timer_period_ms;
			uint32_t n_tables;
		} pipeline_enable;

		struct {
			struct rte_pipeline *p;
		} pipeline_disable;
	};
};

struct thread_msg_rsp {
	int status;
};

/**
 * Main thread
 */
static struct thread_msg_req *
thread_msg_alloc(void)
{
	size_t size = RTE_MAX(sizeof(struct thread_msg_req),
		sizeof(struct thread_msg_rsp));

	return calloc(1, size);
}

static void
thread_msg_free(struct thread_msg_rsp *rsp)
{
	free(rsp);
}

static struct thread_msg_rsp *
thread_msg_send_recv(uint32_t thread_id,
	struct thread_msg_req *req)
{
	struct thread *t = &thread[thread_id];
	struct rte_ring *msgq_req = t->msgq_req;
	struct rte_ring *msgq_rsp = t->msgq_rsp;
	struct thread_msg_rsp *rsp;
	int status;

	/* send */
	do {
		status = rte_ring_sp_enqueue(msgq_req, req);
	} while (status == -ENOBUFS);

	/* recv */
	do {
		status = rte_ring_sc_dequeue(msgq_rsp, (void **) &rsp);
	} while (status != 0);

	return rsp;
}

int
thread_pipeline_enable(uint32_t thread_id,
	const char *pipeline_name)
{
	struct pipeline *p = pipeline_find(pipeline_name);
	struct thread *t;
	struct thread_msg_req *req;
	struct thread_msg_rsp *rsp;
	uint32_t i;
	int status;

	/* Check input params */
	if ((thread_id >= RTE_MAX_LCORE) ||
		(p == NULL) ||
		(p->n_ports_in == 0) ||
		(p->n_ports_out == 0) ||
		(p->n_tables == 0))
		return -1;

	t = &thread[thread_id];
	if ((t->enabled == 0) ||
		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)
		return -1;

	/* Write request */
	req->type = THREAD_REQ_PIPELINE_ENABLE;
	req->pipeline_enable.p = p->p;
	for (i = 0; i < p->n_tables; i++)
		req->pipeline_enable.table[i].a =
			p->table[i].a;
	req->pipeline_enable.msgq_req = p->msgq_req;
	req->pipeline_enable.msgq_rsp = p->msgq_rsp;
	req->pipeline_enable.timer_period_ms = p->timer_period_ms;
	req->pipeline_enable.n_tables = p->n_tables;

	/* Send request and wait for response */
	rsp = thread_msg_send_recv(thread_id, req);

	/* Read response */
	status = rsp->status;

	/* Free response */
	thread_msg_free(rsp);

	/* Request completion */
	if (status)
		return status;

	p->thread_id = thread_id;
	p->enabled = 1;

	return 0;
}

int
thread_pipeline_disable(uint32_t thread_id,
	const char *pipeline_name)
{
	struct pipeline *p = pipeline_find(pipeline_name);
	struct thread *t;
	struct thread_msg_req *req;
	struct thread_msg_rsp *rsp;
	int status;

	/* Check input params */
	if ((thread_id >= RTE_MAX_LCORE) ||
		(p == NULL))
		return -1;

	t = &thread[thread_id];
	if (t->enabled == 0)
		return -1;

	if (p->enabled == 0)
		return 0;

	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)
		return -1;

	/* Write request */
	req->type = THREAD_REQ_PIPELINE_DISABLE;
	req->pipeline_disable.p = p->p;

	/* Send request and wait for response */
	rsp = thread_msg_send_recv(thread_id, req);

	/* Read response */
	status = rsp->status;

	/* Free response */
	thread_msg_free(rsp);

	/* Request completion */
	if (status)
		return status;

	p->enabled = 0;

	return 0;
}

/**
 * Data plane threads: message handling
 */
static inline struct thread_msg_req *
thread_msg_recv(struct rte_ring *msgq_req)
{
	struct thread_msg_req *req;

	int status = rte_ring_sc_dequeue(msgq_req, (void **) &req);

	if (status != 0)
		return NULL;

	return req;
}

static inline void
thread_msg_send(struct rte_ring *msgq_rsp,
	struct thread_msg_rsp *rsp)
{
	int status;

	do {
		status = rte_ring_sp_enqueue(msgq_rsp, rsp);
	} while (status == -ENOBUFS);
}

static struct thread_msg_rsp *
thread_msg_handle_pipeline_enable(struct thread_data *t,
	struct thread_msg_req *req)
{
	struct thread_msg_rsp *rsp = (struct thread_msg_rsp *) req;
	struct pipeline_data *p = &t->pipeline_data[t->n_pipelines];
	uint32_t i;

	/* Request */
	if (t->n_pipelines >= THREAD_PIPELINES_MAX) {
		rsp->status = -1;
		return rsp;
	}

	t->p[t->n_pipelines] = req->pipeline_enable.p;

	p->p = req->pipeline_enable.p;
	for (i = 0; i < req->pipeline_enable.n_tables; i++)
		p->table_data[i].a =
			req->pipeline_enable.table[i].a;

	p->n_tables = req->pipeline_enable.n_tables;

	p->msgq_req = req->pipeline_enable.msgq_req;
	p->msgq_rsp = req->pipeline_enable.msgq_rsp;
	p->timer_period =
		(rte_get_tsc_hz() * req->pipeline_enable.timer_period_ms) / 1000;
	p->time_next = rte_get_tsc_cycles() + p->timer_period;

	t->n_pipelines++;

	/* Response */
	rsp->status = 0;
	return rsp;
}

static struct thread_msg_rsp *
thread_msg_handle_pipeline_disable(struct thread_data *t,
	struct thread_msg_req *req)
{
	struct thread_msg_rsp *rsp = (struct thread_msg_rsp *) req;
	uint32_t n_pipelines = t->n_pipelines;
	struct rte_pipeline *pipeline = req->pipeline_disable.p;
	uint32_t i;

	/* find pipeline */
	for (i = 0; i < n_pipelines; i++) {
		struct pipeline_data *p = &t->pipeline_data[i];

		if (p->p != pipeline)
			continue;

		if (i < n_pipelines - 1) {
			struct rte_pipeline *pipeline_last =
				t->p[n_pipelines - 1];
			struct pipeline_data *p_last =
				&t->pipeline_data[n_pipelines - 1];

			t->p[i] = pipeline_last;
			memcpy(p, p_last, sizeof(*p));
		}

		t->n_pipelines--;

		rsp->status = 0;
		return rsp;
	}

	/* should not get here */
	rsp->status = 0;
	return rsp;
}

static void
thread_msg_handle(struct thread_data *t)
{
	for ( ; ; ) {
		struct thread_msg_req *req;
		struct thread_msg_rsp *rsp;

		req = thread_msg_recv(t->msgq_req);
		if (req == NULL)
			break;

		switch (req->type) {
		case THREAD_REQ_PIPELINE_ENABLE:
			rsp = thread_msg_handle_pipeline_enable(t, req);
			break;

		case THREAD_REQ_PIPELINE_DISABLE:
			rsp = thread_msg_handle_pipeline_disable(t, req);
			break;

		default:
			rsp = (struct thread_msg_rsp *) req;
			rsp->status = -1;
		}

		thread_msg_send(t->msgq_rsp, rsp);
	}
}

/**
 * Main thread & data plane threads: message passing
 */
enum pipeline_req_type {
	/* Port IN */
	PIPELINE_REQ_PORT_IN_STATS_READ,
	PIPELINE_REQ_PORT_IN_ENABLE,
	PIPELINE_REQ_PORT_IN_DISABLE,

	/* Port OUT */
	PIPELINE_REQ_PORT_OUT_STATS_READ,

	/* Table */
	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_TABLE_RULE_TIME_READ,
	PIPELINE_REQ_MAX
};

struct pipeline_msg_req_port_in_stats_read {
	int clear;
};

struct pipeline_msg_req_port_out_stats_read {
	int clear;
};

struct pipeline_msg_req_table_stats_read {
	int clear;
};

struct pipeline_msg_req_table_rule_add {
	struct table_rule_match match;
	struct table_rule_action action;
};

struct pipeline_msg_req_table_rule_add_default {
	struct table_rule_action action;
};

struct pipeline_msg_req_table_rule_add_bulk {
	struct table_rule_list *list;
	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_table_rule_time_read {
	void *data;
};

struct pipeline_msg_req {
	enum pipeline_req_type type;
	uint32_t id; /* Port IN, port OUT or table ID */

	RTE_STD_C11
	union {
		struct pipeline_msg_req_port_in_stats_read port_in_stats_read;
		struct pipeline_msg_req_port_out_stats_read port_out_stats_read;
		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;
		struct pipeline_msg_req_table_rule_time_read table_rule_time_read;
	};
};

struct pipeline_msg_rsp_port_in_stats_read {
	struct rte_pipeline_port_in_stats stats;
};

struct pipeline_msg_rsp_port_out_stats_read {
	struct rte_pipeline_port_out_stats stats;
};

struct pipeline_msg_rsp_table_stats_read {
	struct rte_pipeline_table_stats stats;
};

struct pipeline_msg_rsp_table_rule_add {
	void *data;
};

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_table_rule_time_read {
	uint64_t timestamp;
};

struct pipeline_msg_rsp {
	int status;

	RTE_STD_C11
	union {
		struct pipeline_msg_rsp_port_in_stats_read port_in_stats_read;
		struct pipeline_msg_rsp_port_out_stats_read port_out_stats_read;
		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;
		struct pipeline_msg_rsp_table_rule_time_read table_rule_time_read;
	};
};

/**
 * Main thread
 */
static struct pipeline_msg_req *
pipeline_msg_alloc(void)
{
	size_t size = RTE_MAX(sizeof(struct pipeline_msg_req),
		sizeof(struct pipeline_msg_rsp));

	return calloc(1, size);
}

static void
pipeline_msg_free(struct pipeline_msg_rsp *rsp)
{
	free(rsp);
}

static struct pipeline_msg_rsp *
pipeline_msg_send_recv(struct pipeline *p,
	struct pipeline_msg_req *req)
{
	struct rte_ring *msgq_req = p->msgq_req;
	struct rte_ring *msgq_rsp = p->msgq_rsp;
	struct pipeline_msg_rsp *rsp;
	int status;

	/* send */
	do {
		status = rte_ring_sp_enqueue(msgq_req, req);
	} while (status == -ENOBUFS);

	/* recv */
	do {
		status = rte_ring_sc_dequeue(msgq_rsp, (void **) &rsp);
	} while (status != 0);

	return rsp;
}

int
pipeline_port_in_stats_read(const char *pipeline_name,
	uint32_t port_id,
	struct rte_pipeline_port_in_stats *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) ||
		(stats == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(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)
		return -1;

	/* Write request */
	req->type = PIPELINE_REQ_PORT_IN_STATS_READ;
	req->id = port_id;
	req->port_in_stats_read.clear = clear;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		memcpy(stats, &rsp->port_in_stats_read.stats, sizeof(*stats));

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_port_in_enable(const char *pipeline_name,
	uint32_t port_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) ||
		(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)
		return -1;

	/* Write request */
	req->type = PIPELINE_REQ_PORT_IN_ENABLE;
	req->id = port_id;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_port_in_disable(const char *pipeline_name,
	uint32_t port_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) ||
		(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)
		return -1;

	/* Write request */
	req->type = PIPELINE_REQ_PORT_IN_DISABLE;
	req->id = port_id;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_port_out_stats_read(const char *pipeline_name,
	uint32_t port_id,
	struct rte_pipeline_port_out_stats *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) ||
		(stats == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(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)
		return -1;

	/* Write request */
	req->type = PIPELINE_REQ_PORT_OUT_STATS_READ;
	req->id = port_id;
	req->port_out_stats_read.clear = clear;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		memcpy(stats, &rsp->port_out_stats_read.stats, sizeof(*stats));

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_table_stats_read(const char *pipeline_name,
	uint32_t table_id,
	struct rte_pipeline_table_stats *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) ||
		(stats == 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_stats_read(p->p,
			table_id,
			stats,
			clear);

		return status;
	}

	/* Allocate request */
	req = pipeline_msg_alloc();
	if (req == NULL)
		return -1;

	/* Write request */
	req->type = PIPELINE_REQ_TABLE_STATS_READ;
	req->id = table_id;
	req->table_stats_read.clear = clear;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		memcpy(stats, &rsp->table_stats_read.stats, sizeof(*stats));

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

static int
match_check(struct table_rule_match *match,
	struct pipeline *p,
	uint32_t table_id)
{
	struct table *table;

	if ((match == NULL) ||
		(p == NULL) ||
		(table_id >= p->n_tables))
		return -1;

	table = &p->table[table_id];
	if (match->match_type != table->params.match_type)
		return -1;

	switch (match->match_type) {
	case TABLE_ACL:
	{
		struct table_acl_params *t = &table->params.match.acl;
		struct table_rule_match_acl *r = &match->match.acl;

		if ((r->ip_version && (t->ip_version == 0)) ||
			((r->ip_version == 0) && t->ip_version))
			return -1;

		if (r->ip_version) {
			if ((r->sa_depth > 32) ||
				(r->da_depth > 32))
				return -1;
		} else {
			if ((r->sa_depth > 128) ||
				(r->da_depth > 128))
				return -1;
		}
		return 0;
	}

	case TABLE_ARRAY:
		return 0;

	case TABLE_HASH:
		return 0;

	case TABLE_LPM:
	{
		struct table_lpm_params *t = &table->params.match.lpm;
		struct table_rule_match_lpm *r = &match->match.lpm;

		if ((r->ip_version && (t->key_size != 4)) ||
			((r->ip_version == 0) && (t->key_size != 16)))
			return -1;

		if (r->ip_version) {
			if (r->depth > 32)
				return -1;
		} else {
			if (r->depth > 128)
				return -1;
		}
		return 0;
	}

	case TABLE_STUB:
		return -1;

	default:
		return -1;
	}
}

static int
action_check(struct table_rule_action *action,
	struct pipeline *p,
	uint32_t table_id)
{
	struct table_action_profile *ap;

	if ((action == NULL) ||
		(p == NULL) ||
		(table_id >= p->n_tables))
		return -1;

	ap = p->table[table_id].ap;
	if (action->action_mask != ap->params.action_mask)
		return -1;

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
		if ((action->fwd.action == RTE_PIPELINE_ACTION_PORT) &&
			(action->fwd.id >= p->n_ports_out))
			return -1;

		if ((action->fwd.action == RTE_PIPELINE_ACTION_TABLE) &&
			(action->fwd.id >= p->n_tables))
			return -1;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_MTR)) {
		uint32_t tc_mask0 = (1 << ap->params.mtr.n_tc) - 1;
		uint32_t tc_mask1 = action->mtr.tc_mask;

		if (tc_mask1 != tc_mask0)
			return -1;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TM)) {
		uint32_t n_subports_per_port =
			ap->params.tm.n_subports_per_port;
		uint32_t n_pipes_per_subport =
			ap->params.tm.n_pipes_per_subport;
		uint32_t subport_id = action->tm.subport_id;
		uint32_t pipe_id = action->tm.pipe_id;

		if ((subport_id >= n_subports_per_port) ||
			(pipe_id >= n_pipes_per_subport))
			return -1;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_ENCAP)) {
		uint64_t encap_mask = ap->params.encap.encap_mask;
		enum rte_table_action_encap_type type = action->encap.type;

		if ((encap_mask & (1LLU << type)) == 0)
			return -1;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_NAT)) {
		int ip_version0 = ap->params.common.ip_version;
		int ip_version1 = action->nat.ip_version;

		if ((ip_version1 && (ip_version0 == 0)) ||
			((ip_version1 == 0) && ip_version0))
			return -1;
	}

	return 0;
}

static int
action_default_check(struct table_rule_action *action,
	struct pipeline *p,
	uint32_t table_id)
{
	if ((action == NULL) ||
		(action->action_mask != (1LLU << RTE_TABLE_ACTION_FWD)) ||
		(p == NULL) ||
		(table_id >= p->n_tables))
		return -1;

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
		if ((action->fwd.action == RTE_PIPELINE_ACTION_PORT) &&
			(action->fwd.id >= p->n_ports_out))
			return -1;

		if ((action->fwd.action == RTE_PIPELINE_ACTION_TABLE) &&
			(action->fwd.id >= p->n_tables))
			return -1;
	}

	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);

struct table_ll {
	struct rte_pipeline *p;
	int table_id;
	struct rte_table_action *a;
	int bulk_supported;
};

static int
table_rule_add_bulk_ll(struct table_ll *table,
	struct table_rule_list *list,
	uint32_t *n_rules)
{
	union table_rule_match_low_level *match_ll = NULL;
	uint8_t *action_ll = NULL;
	void **match_ll_ptr = NULL;
	struct rte_pipeline_table_entry **action_ll_ptr = NULL;
	struct rte_pipeline_table_entry **entries_ptr = NULL;
	int *found = NULL;
	struct table_rule *rule;
	uint32_t n, i;
	int status = 0;

	n = 0;
	TAILQ_FOREACH(rule, list, node)
		n++;

	/* Memory allocation */
	match_ll = calloc(n, sizeof(union table_rule_match_low_level));
	action_ll = calloc(n, TABLE_RULE_ACTION_SIZE_MAX);

	match_ll_ptr = calloc(n, sizeof(void *));
	action_ll_ptr = calloc(n, sizeof(struct rte_pipeline_table_entry *));

	entries_ptr = calloc(n, sizeof(struct rte_pipeline_table_entry *));
	found = calloc(n, sizeof(int));

	if (match_ll == NULL ||
		action_ll == NULL ||
		match_ll_ptr == NULL ||
		action_ll_ptr == NULL ||
		entries_ptr == NULL ||
		found == NULL) {
			status = -ENOMEM;
			goto table_rule_add_bulk_ll_free;
	}

	/* Init */
	for (i = 0; i < n; 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, action) conversion */
	i = 0;
	TAILQ_FOREACH(rule, list, node) {
		status = match_convert(&rule->match, match_ll_ptr[i], 1);
		if (status)
			goto table_rule_add_bulk_ll_free;

		status = action_convert(table->a, &rule->action, action_ll_ptr[i]);
		if (status)
			goto table_rule_add_bulk_ll_free;

		i++;
	}

	/* Add rule (match, action) to table */
	if (table->bulk_supported) {
		status = rte_pipeline_table_entry_add_bulk(table->p,
			table->table_id,
			match_ll_ptr,
			action_ll_ptr,
			n,
			found,
			entries_ptr);
		if (status)
			goto table_rule_add_bulk_ll_free;
	} else
		for (i = 0; i < n; i++) {
			status = rte_pipeline_table_entry_add(table->p,
				table->table_id,
				match_ll_ptr[i],
				action_ll_ptr[i],
				&found[i],
				&entries_ptr[i]);
			if (status) {
				if (i == 0)
					goto table_rule_add_bulk_ll_free;

				/* No roll-back. */
				status = 0;
				n = i;
				break;
			}
		}

	/* Write back to the rule list. */
	i = 0;
	TAILQ_FOREACH(rule, list, node) {
		if (i >= n)
			break;

		rule->data = entries_ptr[i];

		i++;
	}

	*n_rules = n;

	/* Free */
table_rule_add_bulk_ll_free:
	free(found);
	free(entries_ptr);
	free(action_ll_ptr);
	free(match_ll_ptr);
	free(action_ll);
	free(match_ll);

	return status;
}

int
pipeline_table_rule_add(const char *pipeline_name,
	uint32_t table_id,
	struct table_rule_match *match,
	struct table_rule_action *action)
{
	struct pipeline *p;
	struct table *table;
	struct pipeline_msg_req *req;
	struct pipeline_msg_rsp *rsp;
	struct table_rule *rule;
	int status;

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(match == NULL) ||
		(action == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(table_id >= p->n_tables) ||
		match_check(match, p, table_id) ||
		action_check(action, p, table_id))
		return -1;

	table = &p->table[table_id];

	rule = calloc(1, sizeof(struct table_rule));
	if (rule == NULL)
		return -1;

	memcpy(&rule->match, match, sizeof(*match));
	memcpy(&rule->action, action, sizeof(*action));

	if (!pipeline_is_running(p)) {
		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) {
			free(rule);
			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);
			free(rule);
			return -1;
		}

		status = action_convert(table->a, action, data_in);
		if (status) {
			free(buffer);
			free(rule);
			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);
			free(rule);
			return -1;
		}

		/* Write Response */
		rule->data = data_out;
		table_rule_add(table, rule);

		free(buffer);
		return 0;
	}

	/* Allocate request */
	req = pipeline_msg_alloc();
	if (req == NULL) {
		free(rule);
		return -1;
	}

	/* Write request */
	req->type = PIPELINE_REQ_TABLE_RULE_ADD;
	req->id = table_id;
	memcpy(&req->table_rule_add.match, match, sizeof(*match));
	memcpy(&req->table_rule_add.action, action, sizeof(*action));

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0) {
		rule->data = rsp->table_rule_add.data;
		table_rule_add(table, rule);
	} else
		free(rule);

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_table_rule_add_default(const char *pipeline_name,
	uint32_t table_id,
	struct table_rule_action *action)
{
	struct pipeline *p;
	struct table *table;
	struct pipeline_msg_req *req;
	struct pipeline_msg_rsp *rsp;
	struct table_rule *rule;
	int status;

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(action == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(table_id >= p->n_tables) ||
		action_default_check(action, p, table_id))
		return -1;

	table = &p->table[table_id];

	rule = calloc(1, sizeof(struct table_rule));
	if (rule == NULL)
		return -1;

	memcpy(&rule->action, action, sizeof(*action));

	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) {
			free(rule);
			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);
			free(rule);
			return -1;
		}

		/* Write Response */
		rule->data = data_out;
		table_rule_default_add(table, rule);

		free(buffer);
		return 0;
	}

	/* Allocate request */
	req = pipeline_msg_alloc();
	if (req == NULL) {
		free(rule);
		return -1;
	}

	/* Write request */
	req->type = PIPELINE_REQ_TABLE_RULE_ADD_DEFAULT;
	req->id = table_id;
	memcpy(&req->table_rule_add_default.action, action, sizeof(*action));

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0) {
		rule->data = rsp->table_rule_add_default.data;
		table_rule_default_add(table, rule);
	} else
		free(rule);

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

static uint32_t
table_rule_list_free(struct table_rule_list *list)
{
	uint32_t n = 0;

	if (!list)
		return 0;

	for ( ; ; ) {
		struct table_rule *rule;

		rule = TAILQ_FIRST(list);
		if (rule == NULL)
			break;

		TAILQ_REMOVE(list, rule, node);
		free(rule);
		n++;
	}

	free(list);
	return n;
}

int
pipeline_table_rule_add_bulk(const char *pipeline_name,
	uint32_t table_id,
	struct table_rule_list *list,
	uint32_t *n_rules_added,
	uint32_t *n_rules_not_added)
{
	struct pipeline *p;
	struct table *table;
	struct pipeline_msg_req *req;
	struct pipeline_msg_rsp *rsp;
	struct table_rule *rule;
	int status = 0;

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(list == NULL) ||
		TAILQ_EMPTY(list) ||
		(n_rules_added == NULL) ||
		(n_rules_not_added == NULL)) {
		table_rule_list_free(list);
		return -EINVAL;
	}

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(table_id >= p->n_tables)) {
		table_rule_list_free(list);
		return -EINVAL;
	}

	table = &p->table[table_id];

	TAILQ_FOREACH(rule, list, node)
		if (match_check(&rule->match, p, table_id) ||
			action_check(&rule->action, p, table_id)) {
			table_rule_list_free(list);
			return -EINVAL;
		}

	if (!pipeline_is_running(p)) {
		struct table_ll table_ll = {
			.p = p->p,
			.table_id = table_id,
			.a = table->a,
			.bulk_supported = table->params.match_type == TABLE_ACL,
		};

		status = table_rule_add_bulk_ll(&table_ll, list, n_rules_added);
		if (status) {
			table_rule_list_free(list);
			return status;
		}

		table_rule_add_bulk(table, list, *n_rules_added);
		*n_rules_not_added = table_rule_list_free(list);
		return 0;
	}

	/* Allocate request */
	req = pipeline_msg_alloc();
	if (req == NULL) {
		table_rule_list_free(list);
		return -ENOMEM;
	}

	/* Write request */
	req->type = PIPELINE_REQ_TABLE_RULE_ADD_BULK;
	req->id = table_id;
	req->table_rule_add_bulk.list = list;
	req->table_rule_add_bulk.bulk = table->params.match_type == TABLE_ACL;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0) {
		*n_rules_added = rsp->table_rule_add_bulk.n_rules;

		table_rule_add_bulk(table, list, *n_rules_added);
		*n_rules_not_added = table_rule_list_free(list);
	} else
		table_rule_list_free(list);


	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_table_rule_delete(const char *pipeline_name,
	uint32_t table_id,
	struct table_rule_match *match)
{
	struct pipeline *p;
	struct table *table;
	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;

	table = &p->table[table_id];

	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);

		if (status == 0)
			table_rule_delete(table, match);

		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);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		table_rule_delete(table, match);

	/* 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 table *table;
	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;

	table = &p->table[table_id];

	if (!pipeline_is_running(p)) {
		status = rte_pipeline_table_default_entry_delete(p->p,
			table_id,
			NULL);

		if (status == 0)
			table_rule_default_delete(table);

		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);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		table_rule_default_delete(table);

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_table_rule_stats_read(const char *pipeline_name,
	uint32_t table_id,
	struct table_rule_match *match,
	struct rte_table_action_stats_counters *stats,
	int clear)
{
	struct pipeline *p;
	struct table *table;
	struct pipeline_msg_req *req;
	struct pipeline_msg_rsp *rsp;
	struct table_rule *rule;
	int status;

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(match == NULL) ||
		(stats == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(table_id >= p->n_tables) ||
		match_check(match, p, table_id))
		return -1;

	table = &p->table[table_id];
	rule = table_rule_find(table, match);
	if (rule == NULL)
		return -1;

	if (!pipeline_is_running(p)) {
		status = rte_table_action_stats_read(table->a,
			rule->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 = rule->data;
	req->table_rule_stats_read.clear = clear;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		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);

	/* 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);

	/* 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,
	struct table_rule_match *match,
	struct rte_table_action_mtr_counters *stats,
	int clear)
{
	struct pipeline *p;
	struct table *table;
	struct pipeline_msg_req *req;
	struct pipeline_msg_rsp *rsp;
	struct table_rule *rule;
	uint32_t tc_mask;
	int status;

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(match == NULL) ||
		(stats == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(table_id >= p->n_tables) ||
		match_check(match, p, table_id))
		return -1;

	table = &p->table[table_id];
	tc_mask = (1 << table->ap->params.mtr.n_tc) - 1;

	rule = table_rule_find(table, match);
	if (rule == NULL)
		return -1;

	if (!pipeline_is_running(p)) {
		status = rte_table_action_meter_read(table->a,
				rule->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 = rule->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);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		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);

	/* 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,
	struct table_rule_match *match,
	struct rte_table_action_ttl_counters *stats,
	int clear)
{
	struct pipeline *p;
	struct table *table;
	struct pipeline_msg_req *req;
	struct pipeline_msg_rsp *rsp;
	struct table_rule *rule;
	int status;

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(match == NULL) ||
		(stats == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(table_id >= p->n_tables) ||
		match_check(match, p, table_id))
		return -1;

	table = &p->table[table_id];
	if (!table->ap->params.ttl.n_packets_enabled)
		return -1;

	rule = table_rule_find(table, match);
	if (rule == NULL)
		return -1;

	if (!pipeline_is_running(p)) {
		status = rte_table_action_ttl_read(table->a,
				rule->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 = rule->data;
	req->table_rule_ttl_read.clear = clear;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		memcpy(stats, &rsp->table_rule_ttl_read.stats, sizeof(*stats));

	/* Free response */
	pipeline_msg_free(rsp);

	return status;
}

int
pipeline_table_rule_time_read(const char *pipeline_name,
	uint32_t table_id,
	struct table_rule_match *match,
	uint64_t *timestamp)
{
	struct pipeline *p;
	struct table *table;
	struct pipeline_msg_req *req;
	struct pipeline_msg_rsp *rsp;
	struct table_rule *rule;
	int status;

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(match == NULL) ||
		(timestamp == NULL))
		return -1;

	p = pipeline_find(pipeline_name);
	if ((p == NULL) ||
		(table_id >= p->n_tables) ||
		match_check(match, p, table_id))
		return -1;

	table = &p->table[table_id];

	rule = table_rule_find(table, match);
	if (rule == NULL)
		return -1;

	if (!pipeline_is_running(p)) {
		status = rte_table_action_time_read(table->a,
				rule->data,
				timestamp);

		return status;
	}

	/* Allocate request */
	req = pipeline_msg_alloc();
	if (req == NULL)
		return -1;

	/* Write request */
	req->type = PIPELINE_REQ_TABLE_RULE_TIME_READ;
	req->id = table_id;
	req->table_rule_time_read.data = rule->data;

	/* Send request and wait for response */
	rsp = pipeline_msg_send_recv(p, req);

	/* Read response */
	status = rsp->status;
	if (status == 0)
		*timestamp = rsp->table_rule_time_read.timestamp;

	/* 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->port_out_stats_read.stats,
		clear);

	return rsp;
}

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)
{
	if (depth > 128)
		return -1;

	switch (depth / 32) {
	case 0:
		depth32[0] = depth;
		depth32[1] = 0;
		depth32[2] = 0;
		depth32[3] = 0;
		return 0;

	case 1:
		depth32[0] = 32;
		depth32[1] = depth - 32;
		depth32[2] = 0;
		depth32[3] = 0;
		return 0;

	case 2:
		depth32[0] = 32;
		depth32[1] = 32;
		depth32[2] = depth - 64;
		depth32[3] = 0;
		return 0;

	case 3:
		depth32[0] = 32;
		depth32[1] = 32;
		depth32[2] = 32;
		depth32[3] = depth - 96;
		return 0;

	case 4:
		depth32[0] = 32;
		depth32[1] = 32;
		depth32[2] = 32;
		depth32[3] = 32;
		return 0;

	default:
		return -1;
	}
}

static int
match_convert(struct table_rule_match *mh,
	union table_rule_match_low_level *ml,
	int add)
{
	memset(ml, 0, sizeof(*ml));

	switch (mh->match_type) {
	case TABLE_ACL:
		if (mh->match.acl.ip_version)
			if (add) {
				ml->acl_add.field_value[0].value.u8 =
					mh->match.acl.proto;
				ml->acl_add.field_value[0].mask_range.u8 =
					mh->match.acl.proto_mask;

				ml->acl_add.field_value[1].value.u32 =
					mh->match.acl.ipv4.sa;
				ml->acl_add.field_value[1].mask_range.u32 =
					mh->match.acl.sa_depth;

				ml->acl_add.field_value[2].value.u32 =
					mh->match.acl.ipv4.da;
				ml->acl_add.field_value[2].mask_range.u32 =
					mh->match.acl.da_depth;

				ml->acl_add.field_value[3].value.u16 =
					mh->match.acl.sp0;
				ml->acl_add.field_value[3].mask_range.u16 =
					mh->match.acl.sp1;

				ml->acl_add.field_value[4].value.u16 =
					mh->match.acl.dp0;
				ml->acl_add.field_value[4].mask_range.u16 =
					mh->match.acl.dp1;

				ml->acl_add.priority =
					(int32_t) mh->match.acl.priority;
			} else {
				ml->acl_delete.field_value[0].value.u8 =
					mh->match.acl.proto;
				ml->acl_delete.field_value[0].mask_range.u8 =
					mh->match.acl.proto_mask;

				ml->acl_delete.field_value[1].value.u32 =
					mh->match.acl.ipv4.sa;
				ml->acl_delete.field_value[1].mask_range.u32 =
					mh->match.acl.sa_depth;

				ml->acl_delete.field_value[2].value.u32 =
					mh->match.acl.ipv4.da;
				ml->acl_delete.field_value[2].mask_range.u32 =
					mh->match.acl.da_depth;

				ml->acl_delete.field_value[3].value.u16 =
					mh->match.acl.sp0;
				ml->acl_delete.field_value[3].mask_range.u16 =
					mh->match.acl.sp1;

				ml->acl_delete.field_value[4].value.u16 =
					mh->match.acl.dp0;
				ml->acl_delete.field_value[4].mask_range.u16 =
					mh->match.acl.dp1;
			}
		else
			if (add) {
				uint32_t *sa32 =
					(uint32_t *) mh->match.acl.ipv6.sa;
				uint32_t *da32 =
					(uint32_t *) mh->match.acl.ipv6.da;
				uint32_t sa32_depth[4], da32_depth[4];
				int status;

				status = match_convert_ipv6_depth(
					mh->match.acl.sa_depth,
					sa32_depth);
				if (status)
					return status;

				status = match_convert_ipv6_depth(
					mh->match.acl.da_depth,
					da32_depth);
				if (status)
					return status;

				ml->acl_add.field_value[0].value.u8 =
					mh->match.acl.proto;
				ml->acl_add.field_value[0].mask_range.u8 =
					mh->match.acl.proto_mask;

				ml->acl_add.field_value[1].value.u32 =
					rte_be_to_cpu_32(sa32[0]);
				ml->acl_add.field_value[1].mask_range.u32 =
					sa32_depth[0];
				ml->acl_add.field_value[2].value.u32 =
					rte_be_to_cpu_32(sa32[1]);
				ml->acl_add.field_value[2].mask_range.u32 =
					sa32_depth[1];
				ml->acl_add.field_value[3].value.u32 =
					rte_be_to_cpu_32(sa32[2]);
				ml->acl_add.field_value[3].mask_range.u32 =
					sa32_depth[2];
				ml->acl_add.field_value[4].value.u32 =
					rte_be_to_cpu_32(sa32[3]);
				ml->acl_add.field_value[4].mask_range.u32 =
					sa32_depth[3];

				ml->acl_add.field_value[5].value.u32 =
					rte_be_to_cpu_32(da32[0]);
				ml->acl_add.field_value[5].mask_range.u32 =
					da32_depth[0];
				ml->acl_add.field_value[6].value.u32 =
					rte_be_to_cpu_32(da32[1]);
				ml->acl_add.field_value[6].mask_range.u32 =
					da32_depth[1];
				ml->acl_add.field_value[7].value.u32 =
					rte_be_to_cpu_32(da32[2]);
				ml->acl_add.field_value[7].mask_range.u32 =
					da32_depth[2];
				ml->acl_add.field_value[8].value.u32 =
					rte_be_to_cpu_32(da32[3]);
				ml->acl_add.field_value[8].mask_range.u32 =
					da32_depth[3];

				ml->acl_add.field_value[9].value.u16 =
					mh->match.acl.sp0;
				ml->acl_add.field_value[9].mask_range.u16 =
					mh->match.acl.sp1;

				ml->acl_add.field_value[10].value.u16 =
					mh->match.acl.dp0;
				ml->acl_add.field_value[10].mask_range.u16 =
					mh->match.acl.dp1;

				ml->acl_add.priority =
					(int32_t) mh->match.acl.priority;
			} else {
				uint32_t *sa32 =
					(uint32_t *) mh->match.acl.ipv6.sa;
				uint32_t *da32 =
					(uint32_t *) mh->match.acl.ipv6.da;
				uint32_t sa32_depth[4], da32_depth[4];
				int status;

				status = match_convert_ipv6_depth(
					mh->match.acl.sa_depth,
					sa32_depth);
				if (status)
					return status;

				status = match_convert_ipv6_depth(
					mh->match.acl.da_depth,
					da32_depth);
				if (status)
					return status;

				ml->acl_delete.field_value[0].value.u8 =
					mh->match.acl.proto;
				ml->acl_delete.field_value[0].mask_range.u8 =
					mh->match.acl.proto_mask;

				ml->acl_delete.field_value[1].value.u32 =
					rte_be_to_cpu_32(sa32[0]);
				ml->acl_delete.field_value[1].mask_range.u32 =
					sa32_depth[0];
				ml->acl_delete.field_value[2].value.u32 =
					rte_be_to_cpu_32(sa32[1]);
				ml->acl_delete.field_value[2].mask_range.u32 =
					sa32_depth[1];
				ml->acl_delete.field_value[3].value.u32 =
					rte_be_to_cpu_32(sa32[2]);
				ml->acl_delete.field_value[3].mask_range.u32 =
					sa32_depth[2];
				ml->acl_delete.field_value[4].value.u32 =
					rte_be_to_cpu_32(sa32[3]);
				ml->acl_delete.field_value[4].mask_range.u32 =
					sa32_depth[3];

				ml->acl_delete.field_value[5].value.u32 =
					rte_be_to_cpu_32(da32[0]);
				ml->acl_delete.field_value[5].mask_range.u32 =
					da32_depth[0];
				ml->acl_delete.field_value[6].value.u32 =
					rte_be_to_cpu_32(da32[1]);
				ml->acl_delete.field_value[6].mask_range.u32 =
					da32_depth[1];
				ml->acl_delete.field_value[7].value.u32 =
					rte_be_to_cpu_32(da32[2]);
				ml->acl_delete.field_value[7].mask_range.u32 =
					da32_depth[2];
				ml->acl_delete.field_value[8].value.u32 =
					rte_be_to_cpu_32(da32[3]);
				ml->acl_delete.field_value[8].mask_range.u32 =
					da32_depth[3];

				ml->acl_delete.field_value[9].value.u16 =
					mh->match.acl.sp0;
				ml->acl_delete.field_value[9].mask_range.u16 =
					mh->match.acl.sp1;

				ml->acl_delete.field_value[10].value.u16 =
					mh->match.acl.dp0;
				ml->acl_delete.field_value[10].mask_range.u16 =
					mh->match.acl.dp1;
			}
		return 0;

	case TABLE_ARRAY:
		ml->array.pos = mh->match.array.pos;
		return 0;

	case TABLE_HASH:
		memcpy(ml->hash, mh->match.hash.key, sizeof(ml->hash));
		return 0;

	case TABLE_LPM:
		if (mh->match.lpm.ip_version) {
			ml->lpm_ipv4.ip = mh->match.lpm.ipv4;
			ml->lpm_ipv4.depth = mh->match.lpm.depth;
		} else {
			memcpy(ml->lpm_ipv6.ip,
				mh->match.lpm.ipv6, sizeof(ml->lpm_ipv6.ip));
			ml->lpm_ipv6.depth = mh->match.lpm.depth;
		}

		return 0;

	default:
		return -1;
	}
}

static int
action_convert(struct rte_table_action *a,
	struct table_rule_action *action,
	struct rte_pipeline_table_entry *data)
{
	int status;

	/* Apply actions */
	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_FWD,
			&action->fwd);

		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,
			RTE_TABLE_ACTION_MTR,
			&action->mtr);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TM)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_TM,
			&action->tm);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_ENCAP)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_ENCAP,
			&action->encap);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_NAT)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_NAT,
			&action->nat);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TTL)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_TTL,
			&action->ttl);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_STATS)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_STATS,
			&action->stats);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TIME)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_TIME,
			&action->time);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_SYM_CRYPTO)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_SYM_CRYPTO,
			&action->sym_crypto);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TAG)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_TAG,
			&action->tag);

		if (status)
			return status;
	}

	if (action->action_mask & (1LLU << RTE_TABLE_ACTION_DECAP)) {
		status = rte_table_action_apply(a,
			data,
			RTE_TABLE_ACTION_DECAP,
			&action->decap);

		if (status)
			return status;
	}

	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,
		data_in,
		&key_found,
		&data_out);
	if (status) {
		rsp->status = -1;
		return rsp;
	}

	/* Write response */
	rsp->status = 0;
	rsp->table_rule_add.data = data_out;

	return rsp;
}

static struct pipeline_msg_rsp *
pipeline_msg_handle_table_rule_add_default(struct pipeline_data *p,
	struct pipeline_msg_req *req)
{
	struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
	struct table_rule_action *action = &req->table_rule_add_default.action;
	struct rte_pipeline_table_entry *data_in, *data_out;
	uint32_t table_id = req->id;
	int status;

	/* Apply actions */
	memset(p->buffer, 0, sizeof(p->buffer));
	data_in = (struct rte_pipeline_table_entry *) p->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) {
		rsp->status = -1;
		return rsp;
	}

	/* Write response */
	rsp->status = 0;
	rsp->table_rule_add_default.data = data_out;

	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_list *list = req->table_rule_add_bulk.list;
	uint32_t bulk = req->table_rule_add_bulk.bulk;

	uint32_t n_rules_added;
	int status;

	struct table_ll table_ll = {
		.p = p->p,
		.table_id = table_id,
		.a = p->table_data[table_id].a,
		.bulk_supported = bulk,
	};

	status = table_rule_add_bulk_ll(&table_ll, list, &n_rules_added);
	if (status) {
		rsp->status = -1;
		rsp->table_rule_add_bulk.n_rules = 0;
		return rsp;
	}

	/* Write response */
	rsp->status = 0;
	rsp->table_rule_add_bulk.n_rules = n_rules_added;
	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 struct pipeline_msg_rsp *
pipeline_msg_handle_table_rule_time_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_time_read.data;
	struct rte_table_action *a = p->table_data[table_id].a;

	rsp->status = rte_table_action_time_read(a,
		data,
		&rsp->table_rule_time_read.timestamp);

	return rsp;
}

static void
pipeline_msg_handle(struct pipeline_data *p)
{
	for ( ; ; ) {
		struct pipeline_msg_req *req;
		struct pipeline_msg_rsp *rsp;

		req = pipeline_msg_recv(p->msgq_req);
		if (req == NULL)
			break;

		switch (req->type) {
		case PIPELINE_REQ_PORT_IN_STATS_READ:
			rsp = pipeline_msg_handle_port_in_stats_read(p, req);
			break;

		case PIPELINE_REQ_PORT_IN_ENABLE:
			rsp = pipeline_msg_handle_port_in_enable(p, req);
			break;

		case PIPELINE_REQ_PORT_IN_DISABLE:
			rsp = pipeline_msg_handle_port_in_disable(p, req);
			break;

		case PIPELINE_REQ_PORT_OUT_STATS_READ:
			rsp = pipeline_msg_handle_port_out_stats_read(p, req);
			break;

		case PIPELINE_REQ_TABLE_STATS_READ:
			rsp = pipeline_msg_handle_table_stats_read(p, req);
			break;

		case PIPELINE_REQ_TABLE_RULE_ADD:
			rsp = pipeline_msg_handle_table_rule_add(p, req);
			break;

		case PIPELINE_REQ_TABLE_RULE_ADD_DEFAULT:
			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;

		case PIPELINE_REQ_TABLE_RULE_TIME_READ:
			rsp = pipeline_msg_handle_table_rule_time_read(p, req);
			break;

		default:
			rsp = (struct pipeline_msg_rsp *) req;
			rsp->status = -1;
		}

		pipeline_msg_send(p->msgq_rsp, rsp);
	}
}

/**
 * Data plane threads: main
 */
int
thread_main(void *arg __rte_unused)
{
	struct thread_data *t;
	uint32_t thread_id, i;

	thread_id = rte_lcore_id();
	t = &thread_data[thread_id];

	/* Dispatch loop */
	for (i = 0; ; i++) {
		uint32_t j;

		/* Data Plane */
		for (j = 0; j < t->n_pipelines; j++)
			rte_pipeline_run(t->p[j]);

		/* Control Plane */
		if ((i & 0xF) == 0) {
			uint64_t time = rte_get_tsc_cycles();
			uint64_t time_next_min = UINT64_MAX;

			if (time < t->time_next_min)
				continue;

			/* Pipeline message queues */
			for (j = 0; j < t->n_pipelines; j++) {
				struct pipeline_data *p =
					&t->pipeline_data[j];
				uint64_t time_next = p->time_next;

				if (time_next <= time) {
					pipeline_msg_handle(p);
					rte_pipeline_flush(p->p);
					time_next = time + p->timer_period;
					p->time_next = time_next;
				}

				if (time_next < time_next_min)
					time_next_min = time_next;
			}

			/* Thread message queues */
			{
				uint64_t time_next = t->time_next;

				if (time_next <= time) {
					thread_msg_handle(t);
					time_next = time + t->timer_period;
					t->time_next = time_next;
				}

				if (time_next < time_next_min)
					time_next_min = time_next;
			}

			t->time_next_min = time_next_min;
		}
	}

	return 0;
}