[dpdk-dev] [PATCH v3 2/3] port: add kni nodrop writer

[Dumitrescu, Cristian]

> -----Original Message-----
> From: WeiJie Zhuang [mailto:zhuangwj at gmail.com]
> Sent: Thursday, June 16, 2016 12:27 PM
> To: Dumitrescu, Cristian <cristian.dumitrescu at intel.com>
> Cc: dev at dpdk.org; Singh, Jasvinder <jasvinder.singh at intel.com>; Yigit,
> Ferruh <ferruh.yigit at intel.com>; WeiJie Zhuang <zhuangwj at gmail.com>
> Subject: [PATCH v3 2/3] port: add kni nodrop writer
> 
> 1. add no drop writing operations to the kni port
> 2. support dropless kni config in the ip pipeline sample application
> 
> Signed-off-by: WeiJie Zhuang <zhuangwj at gmail.com>
> ---
>  examples/ip_pipeline/app.h           |   2 +
>  examples/ip_pipeline/config_parse.c  |  31 ++++-
>  examples/ip_pipeline/init.c          |  26 ++++-
>  examples/ip_pipeline/pipeline_be.h   |   6 +
>  lib/librte_port/rte_port_kni.c       | 220
> +++++++++++++++++++++++++++++++++++
>  lib/librte_port/rte_port_kni.h       |  13 +++
>  lib/librte_port/rte_port_version.map |   1 +
>  7 files changed, 292 insertions(+), 7 deletions(-)
> 
> diff --git a/examples/ip_pipeline/app.h b/examples/ip_pipeline/app.h
> index abbd6d4..6a6fdd9 100644
> --- a/examples/ip_pipeline/app.h
> +++ b/examples/ip_pipeline/app.h
> @@ -147,6 +147,8 @@ struct app_pktq_kni_params {
>  	uint32_t mempool_id; /* Position in the app->mempool_params */
>  	uint32_t burst_read;
>  	uint32_t burst_write;
> +	uint32_t dropless;
> +	uint64_t n_retries;
>  };
> 
>  #ifndef APP_FILE_NAME_SIZE
> diff --git a/examples/ip_pipeline/config_parse.c
> b/examples/ip_pipeline/config_parse.c
> index c55be31..31a50c2 100644
> --- a/examples/ip_pipeline/config_parse.c
> +++ b/examples/ip_pipeline/config_parse.c
> @@ -199,6 +199,8 @@ struct app_pktq_kni_params default_kni_params = {
>  	.mempool_id = 0,
>  	.burst_read = 32,
>  	.burst_write = 32,
> +	.dropless = 0,
> +	.n_retries = 0,
>  };
> 
>  struct app_pktq_source_params default_source_params = {
> @@ -1927,7 +1929,7 @@ parse_kni(struct app_params *app,
> 
>  		if (strcmp(ent->name, "mempool") == 0) {
>  			int status = validate_name(ent->value,
> -
> "MEMPOOL", 1);
> +				"MEMPOOL", 1);
>  			ssize_t idx;
> 
>  			PARSE_ERROR((status == 0), section_name,
> @@ -1940,7 +1942,7 @@ parse_kni(struct app_params *app,
> 
>  		if (strcmp(ent->name, "burst_read") == 0) {
>  			int status = parser_read_uint32(&param-
> >burst_read,
> -
> 		ent->value);
> +						ent->value);
> 
>  			PARSE_ERROR((status == 0), section_name,
>  						ent->name);
> @@ -1949,7 +1951,25 @@ parse_kni(struct app_params *app,
> 
>  		if (strcmp(ent->name, "burst_write") == 0) {
>  			int status = parser_read_uint32(&param-
> >burst_write,
> -
> 		ent->value);
> +						ent->value);
> +
> +			PARSE_ERROR((status == 0), section_name,
> +						ent->name);
> +			continue;
> +		}
> +
> +		if (strcmp(ent->name, "dropless") == 0) {
> +			int status = parser_read_arg_bool(ent->value);
> +
> +			PARSE_ERROR((status != -EINVAL), section_name,
> +						ent->name);
> +			param->dropless = status;
> +			continue;
> +		}
> +
> +		if (strcmp(ent->name, "n_retries") == 0) {
> +			int status = parser_read_uint64(&param->n_retries,
> +						ent->value);
> 
>  			PARSE_ERROR((status == 0), section_name,
>  						ent->name);
> @@ -2794,6 +2814,11 @@ save_kni_params(struct app_params *app, FILE
> *f)
>  		/* burst_write */
>  		fprintf(f, "%s = %" PRIu32 "\n", "burst_write", p-
> >burst_write);
> 
> +		/* dropless */
> +		fprintf(f, "%s = %s\n",
> +				"dropless",
> +				p->dropless ? "yes" : "no");

Please also do not forget to save the number of retries parameter as well (struct app_pktq_kni_params::n_retries):

		fprintf(f, "%s = %" PRIu64 "\n", "n_retries", p->n_retries);

I realize that we forgot to save n_retires in function save_txq_params(), can you please add it as part of your patch? We have it for save_swq_params().

> +
>  		fputc('\n', f);
>  	}
>  }
> diff --git a/examples/ip_pipeline/init.c b/examples/ip_pipeline/init.c
> index d522de4..af24f52 100644
> --- a/examples/ip_pipeline/init.c
> +++ b/examples/ip_pipeline/init.c
> @@ -1434,10 +1434,28 @@ void app_pipeline_params_get(struct
> app_params *app,
>  #ifdef RTE_LIBRTE_KNI
>  		case APP_PKTQ_OUT_KNI:
>  		{
> -			out->type = PIPELINE_PORT_OUT_KNI_WRITER;
> -			out->params.kni.kni = app->kni[in->id];
> -			out->params.kni.tx_burst_sz =
> -				app->kni_params[in->id].burst_write;
> +			struct app_pktq_kni_params *p_kni =
> +				&app->kni_params[in->id];
> +
> +			if (p_kni->dropless == 0) {
> +				struct rte_port_kni_writer_params *params
> =
> +					&out->params.kni;
> +
> +				out->type =
> PIPELINE_PORT_OUT_KNI_WRITER;
> +				params->kni = app->kni[in->id];
> +				params->tx_burst_sz =
> +					app->kni_params[in-
> >id].burst_write;
> +			} else {
> +				struct rte_port_kni_writer_nodrop_params
> +					*params = &out-
> >params.kni_nodrop;
> +
> +				out->type =
> PIPELINE_PORT_OUT_KNI_WRITER_NODROP;
> +				params->kni = app->kni[in->id];
> +				params->tx_burst_sz =
> +					app->kni_params[in-
> >id].burst_write;
> +				params->n_retries =
> +					app->kni_params[in->id].n_retries;
> +			}
>  			break;
>  		}
>  #endif /* RTE_LIBRTE_KNI */
> diff --git a/examples/ip_pipeline/pipeline_be.h
> b/examples/ip_pipeline/pipeline_be.h
> index 0ee208c..b562472 100644
> --- a/examples/ip_pipeline/pipeline_be.h
> +++ b/examples/ip_pipeline/pipeline_be.h
> @@ -138,6 +138,7 @@ enum pipeline_port_out_type {
>  	PIPELINE_PORT_OUT_RING_WRITER_IPV6_RAS,
>  	PIPELINE_PORT_OUT_SCHED_WRITER,
>  	PIPELINE_PORT_OUT_KNI_WRITER,
> +	PIPELINE_PORT_OUT_KNI_WRITER_NODROP,
>  	PIPELINE_PORT_OUT_SINK,
>  };
> 
> @@ -155,6 +156,7 @@ struct pipeline_port_out_params {
>  		struct rte_port_sched_writer_params sched;
>  #ifdef RTE_LIBRTE_KNI
>  		struct rte_port_kni_writer_params kni;
> +		struct rte_port_kni_writer_nodrop_params kni_nodrop;
>  #endif
>  		struct rte_port_sink_params sink;
>  	} params;
> @@ -185,6 +187,8 @@ pipeline_port_out_params_convert(struct
> pipeline_port_out_params  *p)
>  #ifdef RTE_LIBRTE_KNI
>  	case PIPELINE_PORT_OUT_KNI_WRITER:
>  		return (void *) &p->params.kni;
> +	case PIPELINE_PORT_OUT_KNI_WRITER_NODROP:
> +		return (void *) &p->params.kni_nodrop;
>  #endif
>  	case PIPELINE_PORT_OUT_SINK:
>  		return (void *) &p->params.sink;
> @@ -218,6 +222,8 @@ pipeline_port_out_params_get_ops(struct
> pipeline_port_out_params  *p)
>  #ifdef RTE_LIBRTE_KNI
>  	case PIPELINE_PORT_OUT_KNI_WRITER:
>  		return &rte_port_kni_writer_ops;
> +	case PIPELINE_PORT_OUT_KNI_WRITER_NODROP:
> +		return &rte_port_kni_writer_nodrop_ops;
>  #endif
>  	case PIPELINE_PORT_OUT_SINK:
>  		return &rte_port_sink_ops;
> diff --git a/lib/librte_port/rte_port_kni.c b/lib/librte_port/rte_port_kni.c
> index 4cbc345..08f4ac2 100644
> --- a/lib/librte_port/rte_port_kni.c
> +++ b/lib/librte_port/rte_port_kni.c
> @@ -306,6 +306,217 @@ static int rte_port_kni_writer_stats_read(void
> *port,
>  }
> 
>  /*
> + * Port KNI Writer Nodrop
> + */
> +#ifdef RTE_PORT_STATS_COLLECT
> +
> +#define RTE_PORT_KNI_WRITER_NODROP_STATS_PKTS_IN_ADD(port, val)
> \
> +	port->stats.n_pkts_in += val
> +#define RTE_PORT_KNI_WRITER_NODROP_STATS_PKTS_DROP_ADD(port,
> val) \
> +	port->stats.n_pkts_drop += val
> +
> +#else
> +
> +#define RTE_PORT_KNI_WRITER_NODROP_STATS_PKTS_IN_ADD(port, val)
> +#define RTE_PORT_KNI_WRITER_NODROP_STATS_PKTS_DROP_ADD(port,
> val)
> +
> +#endif
> +
> +struct rte_port_kni_writer_nodrop {
> +	struct rte_port_out_stats stats;
> +
> +	struct rte_mbuf *tx_buf[2 * RTE_PORT_IN_BURST_SIZE_MAX];
> +	uint32_t tx_burst_sz;
> +	uint32_t tx_buf_count;
> +	uint64_t bsz_mask;
> +	uint64_t n_retries;
> +	struct rte_kni *kni;
> +};
> +
> +static void *
> +rte_port_kni_writer_nodrop_create(void *params, int socket_id)
> +{
> +	struct rte_port_kni_writer_nodrop_params *conf =
> +		(struct rte_port_kni_writer_nodrop_params *) params;
> +	struct rte_port_kni_writer_nodrop *port;
> +
> +	/* Check input parameters */
> +	if ((conf == NULL) ||
> +		(conf->tx_burst_sz == 0) ||
> +		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX) ||
> +		(!rte_is_power_of_2(conf->tx_burst_sz))) {
> +		RTE_LOG(ERR, PORT, "%s: Invalid input parameters\n",
> __func__);
> +		return NULL;
> +	}
> +
> +	/* Memory allocation */
> +	port = rte_zmalloc_socket("PORT", sizeof(*port),
> +		RTE_CACHE_LINE_SIZE, socket_id);
> +	if (port == NULL) {
> +		RTE_LOG(ERR, PORT, "%s: Failed to allocate port\n",
> __func__);
> +		return NULL;
> +	}
> +
> +	/* Initialization */
> +	port->kni = conf->kni;
> +	port->tx_burst_sz = conf->tx_burst_sz;
> +	port->tx_buf_count = 0;
> +	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
> +
> +	/*
> +	 * When n_retries is 0 it means that we should wait for every packet
> to
> +	 * send no matter how many retries should it take. To limit number of
> +	 * branches in fast path, we use UINT64_MAX instead of branching.
> +	 */
> +	port->n_retries = (conf->n_retries == 0) ? UINT64_MAX : conf-
> >n_retries;
> +
> +	return port;
> +}
> +
> +static inline void
> +send_burst_nodrop(struct rte_port_kni_writer_nodrop *p)
> +{
> +	uint32_t nb_tx = 0, i;
> +
> +	nb_tx = rte_kni_tx_burst(p->kni, p->tx_buf, p->tx_buf_count);
> +
> +	/* We sent all the packets in a first try */
> +	if (nb_tx >= p->tx_buf_count) {
> +		p->tx_buf_count = 0;
> +		return;
> +	}
> +
> +	for (i = 0; i < p->n_retries; i++) {
> +		nb_tx += rte_kni_tx_burst(p->kni,
> +			p->tx_buf + nb_tx,
> +			p->tx_buf_count - nb_tx);
> +
> +		/* We sent all the packets in more than one try */
> +		if (nb_tx >= p->tx_buf_count) {
> +			p->tx_buf_count = 0;
> +			return;
> +		}
> +	}
> +
> +	/* We didn't send the packets in maximum allowed attempts */
> +	RTE_PORT_KNI_WRITER_NODROP_STATS_PKTS_DROP_ADD(p, p-
> >tx_buf_count - nb_tx);
> +	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
> +		rte_pktmbuf_free(p->tx_buf[nb_tx]);
> +
> +	p->tx_buf_count = 0;
> +}
> +
> +static int
> +rte_port_kni_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
> +{
> +	struct rte_port_kni_writer_nodrop *p =
> +			(struct rte_port_kni_writer_nodrop *) port;
> +
> +	p->tx_buf[p->tx_buf_count++] = pkt;
> +	RTE_PORT_KNI_WRITER_STATS_PKTS_IN_ADD(p, 1);
> +	if (p->tx_buf_count >= p->tx_burst_sz)
> +		send_burst_nodrop(p);
> +
> +	return 0;
> +}
> +
> +static int
> +rte_port_kni_writer_nodrop_tx_bulk(void *port,
> +	struct rte_mbuf **pkts,
> +	uint64_t pkts_mask)
> +{
> +	struct rte_port_kni_writer_nodrop *p =
> +			(struct rte_port_kni_writer_nodrop *) port;
> +
> +	uint64_t bsz_mask = p->bsz_mask;
> +	uint32_t tx_buf_count = p->tx_buf_count;
> +	uint64_t expr = (pkts_mask & (pkts_mask + 1)) |
> +					((pkts_mask & bsz_mask) ^
> bsz_mask);
> +
> +	if (expr == 0) {
> +		uint64_t n_pkts = __builtin_popcountll(pkts_mask);
> +		uint32_t n_pkts_ok;
> +
> +		if (tx_buf_count)
> +			send_burst_nodrop(p);
> +
> +		RTE_PORT_KNI_WRITER_NODROP_STATS_PKTS_IN_ADD(p,
> n_pkts);
> +		n_pkts_ok = rte_kni_tx_burst(p->kni, pkts, n_pkts);
> +
> +		if (n_pkts_ok >= n_pkts)
> +			return 0;
> +
> +		/*
> +		 * If we didn't manage to send all packets in single burst,
> move
> +		 * remaining packets to the buffer and call send burst.
> +		 */
> +		for (; n_pkts_ok < n_pkts; n_pkts_ok++) {
> +			struct rte_mbuf *pkt = pkts[n_pkts_ok];
> +			p->tx_buf[p->tx_buf_count++] = pkt;
> +		}
> +		send_burst_nodrop(p);
> +	} else {
> +		for ( ; pkts_mask; ) {
> +			uint32_t pkt_index = __builtin_ctzll(pkts_mask);
> +			uint64_t pkt_mask = 1LLU << pkt_index;
> +			struct rte_mbuf *pkt = pkts[pkt_index];
> +
> +			p->tx_buf[tx_buf_count++] = pkt;
> +
> 	RTE_PORT_KNI_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
> +			pkts_mask &= ~pkt_mask;
> +		}
> +
> +		p->tx_buf_count = tx_buf_count;
> +		if (tx_buf_count >= p->tx_burst_sz)
> +			send_burst_nodrop(p);
> +	}
> +
> +	return 0;
> +}
> +
> +static int
> +rte_port_kni_writer_nodrop_flush(void *port)
> +{
> +	struct rte_port_kni_writer_nodrop *p =
> +		(struct rte_port_kni_writer_nodrop *) port;
> +
> +	if (p->tx_buf_count > 0)
> +		send_burst_nodrop(p);
> +
> +	return 0;
> +}
> +
> +static int
> +rte_port_kni_writer_nodrop_free(void *port)
> +{
> +	if (port == NULL) {
> +		RTE_LOG(ERR, PORT, "%s: Port is NULL\n", __func__);
> +		return -EINVAL;
> +	}
> +
> +	rte_port_kni_writer_nodrop_flush(port);
> +	rte_free(port);
> +
> +	return 0;
> +}
> +
> +static int rte_port_kni_writer_nodrop_stats_read(void *port,
> +	struct rte_port_out_stats *stats, int clear)
> +{
> +	struct rte_port_kni_writer_nodrop *p =
> +			(struct rte_port_kni_writer_nodrop *) port;
> +
> +	if (stats != NULL)
> +		memcpy(stats, &p->stats, sizeof(p->stats));
> +
> +	if (clear)
> +		memset(&p->stats, 0, sizeof(p->stats));
> +
> +	return 0;
> +}
> +
> +
> +/*
>   * Summary of port operations
>   */
>  struct rte_port_in_ops rte_port_kni_reader_ops = {
> @@ -323,3 +534,12 @@ struct rte_port_out_ops rte_port_kni_writer_ops =
> {
>  	.f_flush = rte_port_kni_writer_flush,
>  	.f_stats = rte_port_kni_writer_stats_read,
>  };
> +
> +struct rte_port_out_ops rte_port_kni_writer_nodrop_ops = {
> +	.f_create = rte_port_kni_writer_nodrop_create,
> +	.f_free = rte_port_kni_writer_nodrop_free,
> +	.f_tx = rte_port_kni_writer_nodrop_tx,
> +	.f_tx_bulk = rte_port_kni_writer_nodrop_tx_bulk,
> +	.f_flush = rte_port_kni_writer_nodrop_flush,
> +	.f_stats = rte_port_kni_writer_nodrop_stats_read,
> +};
> diff --git a/lib/librte_port/rte_port_kni.h b/lib/librte_port/rte_port_kni.h
> index d4de8c4..4b60689 100644
> --- a/lib/librte_port/rte_port_kni.h
> +++ b/lib/librte_port/rte_port_kni.h
> @@ -75,6 +75,19 @@ struct rte_port_kni_writer_params {
>  /** kni_writer port operations */
>  extern struct rte_port_out_ops rte_port_kni_writer_ops;
> 
> +/** kni_writer_nodrop port parameters */
> +struct rte_port_kni_writer_nodrop_params {
> +	/** KNI interface reference */
> +	struct rte_kni *kni;
> +	/** Burst size to KNI interface. */
> +	uint32_t tx_burst_sz;
> +	/** Maximum number of retries, 0 for no limit */
> +	uint32_t n_retries;
> +};
> +
> +/** kni_writer_nodrop port operations */
> +extern struct rte_port_out_ops rte_port_kni_writer_nodrop_ops;
> +
>  #ifdef __cplusplus
>  }
>  #endif
> diff --git a/lib/librte_port/rte_port_version.map
> b/lib/librte_port/rte_port_version.map
> index e61b3fa..c528658 100644
> --- a/lib/librte_port/rte_port_version.map
> +++ b/lib/librte_port/rte_port_version.map
> @@ -41,5 +41,6 @@ DPDK_16.07 {
> 
>  	rte_port_kni_reader_ops;
>  	rte_port_kni_writer_ops;
> +	rte_port_kni_writer_nodrop_ops;
> 
>  } DPDK_2.2;
> --
> 2.7.4