numam-dpdk/examples/ip_pipeline/pipeline.c

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

#include <stdlib.h>
#include <string.h>

#include <rte_common.h>
#include <rte_ip.h>
#include <rte_tcp.h>

#include <rte_string_fns.h>
#include <rte_port_ethdev.h>
#ifdef RTE_LIBRTE_KNI
#include <rte_port_kni.h>
#endif
#include <rte_port_ring.h>
#include <rte_port_source_sink.h>
#include <rte_port_fd.h>
#include <rte_port_sched.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 <rte_table_stub.h>

#ifdef RTE_LIBRTE_KNI
#include "kni.h"
#endif
#include "link.h"
#include "mempool.h"
#include "pipeline.h"
#include "tap.h"
#include "tmgr.h"
#include "swq.h"

#include "hash_func.h"

#ifndef PIPELINE_MSGQ_SIZE
#define PIPELINE_MSGQ_SIZE                                 64
#endif

#ifndef TABLE_LPM_NUMBER_TBL8
#define TABLE_LPM_NUMBER_TBL8                              256
#endif

static struct pipeline_list pipeline_list;

int
pipeline_init(void)
{
	TAILQ_INIT(&pipeline_list);

	return 0;
}

struct pipeline *
pipeline_find(const char *name)
{
	struct pipeline *pipeline;

	if (name == NULL)
		return NULL;

	TAILQ_FOREACH(pipeline, &pipeline_list, node)
		if (strcmp(name, pipeline->name) == 0)
			return pipeline;

	return NULL;
}

struct pipeline *
pipeline_create(const char *name, struct pipeline_params *params)
{
	char msgq_name[NAME_MAX];
	struct rte_pipeline_params pp;
	struct pipeline *pipeline;
	struct rte_pipeline *p;
	struct rte_ring *msgq_req;
	struct rte_ring *msgq_rsp;

	/* Check input params */
	if ((name == NULL) ||
		pipeline_find(name) ||
		(params == NULL) ||
		(params->timer_period_ms == 0))
		return NULL;

	/* Resource create */
	snprintf(msgq_name, sizeof(msgq_name), "%s-MSGQ-REQ", name);

	msgq_req = rte_ring_create(msgq_name,
		PIPELINE_MSGQ_SIZE,
		params->cpu_id,
		RING_F_SP_ENQ | RING_F_SC_DEQ);
	if (msgq_req == NULL)
		return NULL;

	snprintf(msgq_name, sizeof(msgq_name), "%s-MSGQ-RSP", name);

	msgq_rsp = rte_ring_create(msgq_name,
		PIPELINE_MSGQ_SIZE,
		params->cpu_id,
		RING_F_SP_ENQ | RING_F_SC_DEQ);
	if (msgq_rsp == NULL) {
		rte_ring_free(msgq_req);
		return NULL;
	}

	pp.name = name;
	pp.socket_id = (int) params->cpu_id;
	pp.offset_port_id = params->offset_port_id;

	p = rte_pipeline_create(&pp);
	if (p == NULL) {
		rte_ring_free(msgq_rsp);
		rte_ring_free(msgq_req);
		return NULL;
	}

	/* Node allocation */
	pipeline = calloc(1, sizeof(struct pipeline));
	if (pipeline == NULL) {
		rte_pipeline_free(p);
		rte_ring_free(msgq_rsp);
		rte_ring_free(msgq_req);
		return NULL;
	}

	/* Node fill in */
	strlcpy(pipeline->name, name, sizeof(pipeline->name));
	pipeline->p = p;
	pipeline->n_ports_in = 0;
	pipeline->n_ports_out = 0;
	pipeline->n_tables = 0;
	pipeline->msgq_req = msgq_req;
	pipeline->msgq_rsp = msgq_rsp;
	pipeline->timer_period_ms = params->timer_period_ms;
	pipeline->enabled = 0;
	pipeline->cpu_id = params->cpu_id;

	/* Node add to list */
	TAILQ_INSERT_TAIL(&pipeline_list, pipeline, node);

	return pipeline;
}

int
pipeline_port_in_create(const char *pipeline_name,
	struct port_in_params *params,
	int enabled)
{
	struct rte_pipeline_port_in_params p;

	union {
		struct rte_port_ethdev_reader_params ethdev;
		struct rte_port_ring_reader_params ring;
		struct rte_port_sched_reader_params sched;
		struct rte_port_fd_reader_params fd;
#ifdef RTE_LIBRTE_KNI
		struct rte_port_kni_reader_params kni;
#endif
		struct rte_port_source_params source;
	} pp;

	struct pipeline *pipeline;
	struct port_in *port_in;
	struct port_in_action_profile *ap;
	struct rte_port_in_action *action;
	uint32_t port_id;
	int status;

	memset(&p, 0, sizeof(p));
	memset(&pp, 0, sizeof(pp));

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(params == NULL) ||
		(params->burst_size == 0) ||
		(params->burst_size > RTE_PORT_IN_BURST_SIZE_MAX))
		return -1;

	pipeline = pipeline_find(pipeline_name);
	if (pipeline == NULL)
		return -1;

	ap = NULL;
	if (params->action_profile_name) {
		ap = port_in_action_profile_find(params->action_profile_name);
		if (ap == NULL)
			return -1;
	}

	switch (params->type) {
	case PORT_IN_RXQ:
	{
		struct link *link;

		link = link_find(params->dev_name);
		if (link == NULL)
			return -1;

		if (params->rxq.queue_id >= link->n_rxq)
			return -1;

		pp.ethdev.port_id = link->port_id;
		pp.ethdev.queue_id = params->rxq.queue_id;

		p.ops = &rte_port_ethdev_reader_ops;
		p.arg_create = &pp.ethdev;
		break;
	}

	case PORT_IN_SWQ:
	{
		struct swq *swq;

		swq = swq_find(params->dev_name);
		if (swq == NULL)
			return -1;

		pp.ring.ring = swq->r;

		p.ops = &rte_port_ring_reader_ops;
		p.arg_create = &pp.ring;
		break;
	}

	case PORT_IN_TMGR:
	{
		struct tmgr_port *tmgr_port;

		tmgr_port = tmgr_port_find(params->dev_name);
		if (tmgr_port == NULL)
			return -1;

		pp.sched.sched = tmgr_port->s;

		p.ops = &rte_port_sched_reader_ops;
		p.arg_create = &pp.sched;
		break;
	}

	case PORT_IN_TAP:
	{
		struct tap *tap;
		struct mempool *mempool;

		tap = tap_find(params->dev_name);
		mempool = mempool_find(params->tap.mempool_name);
		if ((tap == NULL) || (mempool == NULL))
			return -1;

		pp.fd.fd = tap->fd;
		pp.fd.mempool = mempool->m;
		pp.fd.mtu = params->tap.mtu;

		p.ops = &rte_port_fd_reader_ops;
		p.arg_create = &pp.fd;
		break;
	}

#ifdef RTE_LIBRTE_KNI
	case PORT_IN_KNI:
	{
		struct kni *kni;

		kni = kni_find(params->dev_name);
		if (kni == NULL)
			return -1;

		pp.kni.kni = kni->k;

		p.ops = &rte_port_kni_reader_ops;
		p.arg_create = &pp.kni;
		break;
	}
#endif

	case PORT_IN_SOURCE:
	{
		struct mempool *mempool;

		mempool = mempool_find(params->source.mempool_name);
		if (mempool == NULL)
			return -1;

		pp.source.mempool = mempool->m;
		pp.source.file_name = params->source.file_name;
		pp.source.n_bytes_per_pkt = params->source.n_bytes_per_pkt;

		p.ops = &rte_port_source_ops;
		p.arg_create = &pp.source;
		break;
	}

	default:
		return -1;
	}

	p.burst_size = params->burst_size;

	/* Resource create */
	action = NULL;
	p.f_action = NULL;
	p.arg_ah = NULL;

	if (ap) {
		action = rte_port_in_action_create(ap->ap,
			pipeline->cpu_id);
		if (action == NULL)
			return -1;

		status = rte_port_in_action_params_get(
			action,
			&p);
		if (status) {
			rte_port_in_action_free(action);
			return -1;
		}
	}

	status = rte_pipeline_port_in_create(pipeline->p,
		&p,
		&port_id);
	if (status) {
		rte_port_in_action_free(action);
		return -1;
	}

	if (enabled)
		rte_pipeline_port_in_enable(pipeline->p, port_id);

	/* Pipeline */
	port_in = &pipeline->port_in[pipeline->n_ports_in];
	memcpy(&port_in->params, params, sizeof(*params));
	port_in->ap = ap;
	port_in->a = action;
	pipeline->n_ports_in++;

	return 0;
}

int
pipeline_port_in_connect_to_table(const char *pipeline_name,
	uint32_t port_id,
	uint32_t table_id)
{
	struct pipeline *pipeline;
	int status;

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

	pipeline = pipeline_find(pipeline_name);
	if ((pipeline == NULL) ||
		(port_id >= pipeline->n_ports_in) ||
		(table_id >= pipeline->n_tables))
		return -1;

	/* Resource */
	status = rte_pipeline_port_in_connect_to_table(pipeline->p,
		port_id,
		table_id);

	return status;

}

int
pipeline_port_out_create(const char *pipeline_name,
	struct port_out_params *params)
{
	struct rte_pipeline_port_out_params p;

	union {
		struct rte_port_ethdev_writer_params ethdev;
		struct rte_port_ring_writer_params ring;
		struct rte_port_sched_writer_params sched;
		struct rte_port_fd_writer_params fd;
#ifdef RTE_LIBRTE_KNI
		struct rte_port_kni_writer_params kni;
#endif
		struct rte_port_sink_params sink;
	} pp;

	union {
		struct rte_port_ethdev_writer_nodrop_params ethdev;
		struct rte_port_ring_writer_nodrop_params ring;
		struct rte_port_fd_writer_nodrop_params fd;
#ifdef RTE_LIBRTE_KNI
		struct rte_port_kni_writer_nodrop_params kni;
#endif
	} pp_nodrop;

	struct pipeline *pipeline;
	uint32_t port_id;
	int status;

	memset(&p, 0, sizeof(p));
	memset(&pp, 0, sizeof(pp));
	memset(&pp_nodrop, 0, sizeof(pp_nodrop));

	/* Check input params */
	if ((pipeline_name == NULL) ||
		(params == NULL) ||
		(params->burst_size == 0) ||
		(params->burst_size > RTE_PORT_IN_BURST_SIZE_MAX))
		return -1;

	pipeline = pipeline_find(pipeline_name);
	if (pipeline == NULL)
		return -1;

	switch (params->type) {
	case PORT_OUT_TXQ:
	{
		struct link *link;

		link = link_find(params->dev_name);
		if (link == NULL)
			return -1;

		if (params->txq.queue_id >= link->n_txq)
			return -1;

		pp.ethdev.port_id = link->port_id;
		pp.ethdev.queue_id = params->txq.queue_id;
		pp.ethdev.tx_burst_sz = params->burst_size;

		pp_nodrop.ethdev.port_id = link->port_id;
		pp_nodrop.ethdev.queue_id = params->txq.queue_id;
		pp_nodrop.ethdev.tx_burst_sz = params->burst_size;
		pp_nodrop.ethdev.n_retries = params->n_retries;

		if (params->retry == 0) {
			p.ops = &rte_port_ethdev_writer_ops;
			p.arg_create = &pp.ethdev;
		} else {
			p.ops = &rte_port_ethdev_writer_nodrop_ops;
			p.arg_create = &pp_nodrop.ethdev;
		}
		break;
	}

	case PORT_OUT_SWQ:
	{
		struct swq *swq;

		swq = swq_find(params->dev_name);
		if (swq == NULL)
			return -1;

		pp.ring.ring = swq->r;
		pp.ring.tx_burst_sz = params->burst_size;

		pp_nodrop.ring.ring = swq->r;
		pp_nodrop.ring.tx_burst_sz = params->burst_size;
		pp_nodrop.ring.n_retries = params->n_retries;

		if (params->retry == 0) {
			p.ops = &rte_port_ring_writer_ops;
			p.arg_create = &pp.ring;
		} else {
			p.ops = &rte_port_ring_writer_nodrop_ops;
			p.arg_create = &pp_nodrop.ring;
		}
		break;
	}

	case PORT_OUT_TMGR:
	{
		struct tmgr_port *tmgr_port;

		tmgr_port = tmgr_port_find(params->dev_name);
		if (tmgr_port == NULL)
			return -1;

		pp.sched.sched = tmgr_port->s;
		pp.sched.tx_burst_sz = params->burst_size;

		p.ops = &rte_port_sched_writer_ops;
		p.arg_create = &pp.sched;
		break;
	}

	case PORT_OUT_TAP:
	{
		struct tap *tap;

		tap = tap_find(params->dev_name);
		if (tap == NULL)
			return -1;

		pp.fd.fd = tap->fd;
		pp.fd.tx_burst_sz = params->burst_size;

		pp_nodrop.fd.fd = tap->fd;
		pp_nodrop.fd.tx_burst_sz = params->burst_size;
		pp_nodrop.fd.n_retries = params->n_retries;

		if (params->retry == 0) {
			p.ops = &rte_port_fd_writer_ops;
			p.arg_create = &pp.fd;
		} else {
			p.ops = &rte_port_fd_writer_nodrop_ops;
			p.arg_create = &pp_nodrop.fd;
		}
		break;
	}

#ifdef RTE_LIBRTE_KNI
	case PORT_OUT_KNI:
	{
		struct kni *kni;

		kni = kni_find(params->dev_name);
		if (kni == NULL)
			return -1;

		pp.kni.kni = kni->k;
		pp.kni.tx_burst_sz = params->burst_size;

		pp_nodrop.kni.kni = kni->k;
		pp_nodrop.kni.tx_burst_sz = params->burst_size;
		pp_nodrop.kni.n_retries = params->n_retries;

		if (params->retry == 0) {
			p.ops = &rte_port_kni_writer_ops;
			p.arg_create = &pp.kni;
		} else {
			p.ops = &rte_port_kni_writer_nodrop_ops;
			p.arg_create = &pp_nodrop.kni;
		}
		break;
	}
#endif

	case PORT_OUT_SINK:
	{
		pp.sink.file_name = params->sink.file_name;
		pp.sink.max_n_pkts = params->sink.max_n_pkts;

		p.ops = &rte_port_sink_ops;
		p.arg_create = &pp.sink;
		break;
	}

	default:
		return -1;
	}

	p.f_action = NULL;
	p.arg_ah = NULL;

	/* Resource create */
	status = rte_pipeline_port_out_create(pipeline->p,
		&p,
		&port_id);

	if (status)
		return -1;

	/* Pipeline */
	pipeline->n_ports_out++;

	return 0;
}

static const struct rte_acl_field_def table_acl_field_format_ipv4[] = {
	/* Protocol */
	[0] = {
		.type = RTE_ACL_FIELD_TYPE_BITMASK,
		.size = sizeof(uint8_t),
		.field_index = 0,
		.input_index = 0,
		.offset = offsetof(struct ipv4_hdr, next_proto_id),
	},

	/* Source IP address (IPv4) */
	[1] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 1,
		.input_index = 1,
		.offset = offsetof(struct ipv4_hdr, src_addr),
	},

	/* Destination IP address (IPv4) */
	[2] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 2,
		.input_index = 2,
		.offset = offsetof(struct ipv4_hdr, dst_addr),
	},

	/* Source Port */
	[3] = {
		.type = RTE_ACL_FIELD_TYPE_RANGE,
		.size = sizeof(uint16_t),
		.field_index = 3,
		.input_index = 3,
		.offset = sizeof(struct ipv4_hdr) +
			offsetof(struct tcp_hdr, src_port),
	},

	/* Destination Port */
	[4] = {
		.type = RTE_ACL_FIELD_TYPE_RANGE,
		.size = sizeof(uint16_t),
		.field_index = 4,
		.input_index = 3,
		.offset = sizeof(struct ipv4_hdr) +
			offsetof(struct tcp_hdr, dst_port),
	},
};

static const struct rte_acl_field_def table_acl_field_format_ipv6[] = {
	/* Protocol */
	[0] = {
		.type = RTE_ACL_FIELD_TYPE_BITMASK,
		.size = sizeof(uint8_t),
		.field_index = 0,
		.input_index = 0,
		.offset = offsetof(struct ipv6_hdr, proto),
	},

	/* Source IP address (IPv6) */
	[1] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 1,
		.input_index = 1,
		.offset = offsetof(struct ipv6_hdr, src_addr[0]),
	},

	[2] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 2,
		.input_index = 2,
		.offset = offsetof(struct ipv6_hdr, src_addr[4]),
	},

	[3] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 3,
		.input_index = 3,
		.offset = offsetof(struct ipv6_hdr, src_addr[8]),
	},

	[4] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 4,
		.input_index = 4,
		.offset = offsetof(struct ipv6_hdr, src_addr[12]),
	},

	/* Destination IP address (IPv6) */
	[5] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 5,
		.input_index = 5,
		.offset = offsetof(struct ipv6_hdr, dst_addr[0]),
	},

	[6] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 6,
		.input_index = 6,
		.offset = offsetof(struct ipv6_hdr, dst_addr[4]),
	},

	[7] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 7,
		.input_index = 7,
		.offset = offsetof(struct ipv6_hdr, dst_addr[8]),
	},

	[8] = {
		.type = RTE_ACL_FIELD_TYPE_MASK,
		.size = sizeof(uint32_t),
		.field_index = 8,
		.input_index = 8,
		.offset = offsetof(struct ipv6_hdr, dst_addr[12]),
	},

	/* Source Port */
	[9] = {
		.type = RTE_ACL_FIELD_TYPE_RANGE,
		.size = sizeof(uint16_t),
		.field_index = 9,
		.input_index = 9,
		.offset = sizeof(struct ipv6_hdr) +
			offsetof(struct tcp_hdr, src_port),
	},

	/* Destination Port */
	[10] = {
		.type = RTE_ACL_FIELD_TYPE_RANGE,
		.size = sizeof(uint16_t),
		.field_index = 10,
		.input_index = 9,
		.offset = sizeof(struct ipv6_hdr) +
			offsetof(struct tcp_hdr, dst_port),
	},
};

int
pipeline_table_create(const char *pipeline_name,
	struct table_params *params)
{
	char name[NAME_MAX];
	struct rte_pipeline_table_params p;

	union {
		struct rte_table_acl_params acl;
		struct rte_table_array_params array;
		struct rte_table_hash_params hash;
		struct rte_table_lpm_params lpm;
		struct rte_table_lpm_ipv6_params lpm_ipv6;
	} pp;

	struct pipeline *pipeline;
	struct table *table;
	struct table_action_profile *ap;
	struct rte_table_action *action;
	uint32_t table_id;
	int status;

	memset(&p, 0, sizeof(p));
	memset(&pp, 0, sizeof(pp));

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

	pipeline = pipeline_find(pipeline_name);
	if ((pipeline == NULL) ||
		(pipeline->n_tables >= RTE_PIPELINE_TABLE_MAX))
		return -1;

	ap = NULL;
	if (params->action_profile_name) {
		ap = table_action_profile_find(params->action_profile_name);
		if (ap == NULL)
			return -1;
	}

	snprintf(name, NAME_MAX, "%s_table%u",
		pipeline_name, pipeline->n_tables);

	switch (params->match_type) {
	case TABLE_ACL:
	{
		uint32_t ip_header_offset = params->match.acl.ip_header_offset -
			(sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM);
		uint32_t i;

		if (params->match.acl.n_rules == 0)
			return -1;

		pp.acl.name = name;
		pp.acl.n_rules = params->match.acl.n_rules;
		if (params->match.acl.ip_version) {
			memcpy(&pp.acl.field_format,
				&table_acl_field_format_ipv4,
				sizeof(table_acl_field_format_ipv4));
			pp.acl.n_rule_fields =
				RTE_DIM(table_acl_field_format_ipv4);
		} else {
			memcpy(&pp.acl.field_format,
				&table_acl_field_format_ipv6,
				sizeof(table_acl_field_format_ipv6));
			pp.acl.n_rule_fields =
				RTE_DIM(table_acl_field_format_ipv6);
		}

		for (i = 0; i < pp.acl.n_rule_fields; i++)
			pp.acl.field_format[i].offset += ip_header_offset;

		p.ops = &rte_table_acl_ops;
		p.arg_create = &pp.acl;
		break;
	}

	case TABLE_ARRAY:
	{
		if (params->match.array.n_keys == 0)
			return -1;

		pp.array.n_entries = params->match.array.n_keys;
		pp.array.offset = params->match.array.key_offset;

		p.ops = &rte_table_array_ops;
		p.arg_create = &pp.array;
		break;
	}

	case TABLE_HASH:
	{
		struct rte_table_ops *ops;
		rte_table_hash_op_hash f_hash;

		if (params->match.hash.n_keys == 0)
			return -1;

		switch (params->match.hash.key_size) {
		case  8:
			f_hash = hash_default_key8;
			break;
		case 16:
			f_hash = hash_default_key16;
			break;
		case 24:
			f_hash = hash_default_key24;
			break;
		case 32:
			f_hash = hash_default_key32;
			break;
		case 40:
			f_hash = hash_default_key40;
			break;
		case 48:
			f_hash = hash_default_key48;
			break;
		case 56:
			f_hash = hash_default_key56;
			break;
		case 64:
			f_hash = hash_default_key64;
			break;
		default:
			return -1;
		}

		pp.hash.name = name;
		pp.hash.key_size = params->match.hash.key_size;
		pp.hash.key_offset = params->match.hash.key_offset;
		pp.hash.key_mask = params->match.hash.key_mask;
		pp.hash.n_keys = params->match.hash.n_keys;
		pp.hash.n_buckets = params->match.hash.n_buckets;
		pp.hash.f_hash = f_hash;
		pp.hash.seed = 0;

		if (params->match.hash.extendable_bucket)
			switch (params->match.hash.key_size) {
			case  8:
				ops = &rte_table_hash_key8_ext_ops;
				break;
			case 16:
				ops = &rte_table_hash_key16_ext_ops;
				break;
			default:
				ops = &rte_table_hash_ext_ops;
			}
		else
			switch (params->match.hash.key_size) {
			case  8:
				ops = &rte_table_hash_key8_lru_ops;
				break;
			case 16:
				ops = &rte_table_hash_key16_lru_ops;
				break;
			default:
				ops = &rte_table_hash_lru_ops;
			}

		p.ops = ops;
		p.arg_create = &pp.hash;
		break;
	}

	case TABLE_LPM:
	{
		if (params->match.lpm.n_rules == 0)
			return -1;

		switch (params->match.lpm.key_size) {
		case 4:
		{
			pp.lpm.name = name;
			pp.lpm.n_rules = params->match.lpm.n_rules;
			pp.lpm.number_tbl8s = TABLE_LPM_NUMBER_TBL8;
			pp.lpm.flags = 0;
			pp.lpm.entry_unique_size = p.action_data_size +
				sizeof(struct rte_pipeline_table_entry);
			pp.lpm.offset = params->match.lpm.key_offset;

			p.ops = &rte_table_lpm_ops;
			p.arg_create = &pp.lpm;
			break;
		}

		case 16:
		{
			pp.lpm_ipv6.name = name;
			pp.lpm_ipv6.n_rules = params->match.lpm.n_rules;
			pp.lpm_ipv6.number_tbl8s = TABLE_LPM_NUMBER_TBL8;
			pp.lpm_ipv6.entry_unique_size = p.action_data_size +
				sizeof(struct rte_pipeline_table_entry);
			pp.lpm_ipv6.offset = params->match.lpm.key_offset;

			p.ops = &rte_table_lpm_ipv6_ops;
			p.arg_create = &pp.lpm_ipv6;
			break;
		}

		default:
			return -1;
		}

		break;
	}

	case TABLE_STUB:
	{
		p.ops = &rte_table_stub_ops;
		p.arg_create = NULL;
		break;
	}

	default:
		return -1;
	}

	/* Resource create */
	action = NULL;
	p.f_action_hit = NULL;
	p.f_action_miss = NULL;
	p.arg_ah = NULL;

	if (ap) {
		action = rte_table_action_create(ap->ap,
			pipeline->cpu_id);
		if (action == NULL)
			return -1;

		status = rte_table_action_table_params_get(
			action,
			&p);
		if (status ||
			((p.action_data_size +
			sizeof(struct rte_pipeline_table_entry)) >
			TABLE_RULE_ACTION_SIZE_MAX)) {
			rte_table_action_free(action);
			return -1;
		}
	}

	if (params->match_type == TABLE_LPM) {
		if (params->match.lpm.key_size == 4)
			pp.lpm.entry_unique_size = p.action_data_size +
				sizeof(struct rte_pipeline_table_entry);

		if (params->match.lpm.key_size == 16)
			pp.lpm_ipv6.entry_unique_size = p.action_data_size +
				sizeof(struct rte_pipeline_table_entry);
	}

	status = rte_pipeline_table_create(pipeline->p,
		&p,
		&table_id);
	if (status) {
		rte_table_action_free(action);
		return -1;
	}

	/* Pipeline */
	table = &pipeline->table[pipeline->n_tables];
	memcpy(&table->params, params, sizeof(*params));
	table->ap = ap;
	table->a = action;
	pipeline->n_tables++;

	return 0;
}