Function rte_distributor_process

dpdk/lib/distributor/rte_distributor.c:441–608

process a set of packets to distribute them to workers */

/* process a set of packets to distribute them to workers */
int
rte_distributor_process(struct rte_distributor *d,
		struct rte_mbuf **mbufs, unsigned int num_mbufs)
{
	unsigned int next_idx = 0;
	static unsigned int wkr;
	struct rte_mbuf *next_mb = NULL;
	int64_t next_value = 0;
	uint16_t new_tag = 0;
	uint16_t flows[RTE_DIST_BURST_SIZE] __rte_cache_aligned;
	unsigned int i, j, w, wid, matching_required;

	if (d->alg_type == RTE_DIST_ALG_SINGLE) {
		/* Call the old API */
		return rte_distributor_process_single(d->d_single,
			mbufs, num_mbufs);
	}

	for (wid = 0 ; wid < d->num_workers; wid++)
		handle_returns(d, wid);

	if (unlikely(num_mbufs == 0)) {
		/* Flush out all non-full cache-lines to workers. */
		for (wid = 0 ; wid < d->num_workers; wid++) {
			/* Sync with worker on GET_BUF flag. */
			if (rte_atomic_load_explicit(&(d->bufs[wid].bufptr64[0]),
				rte_memory_order_acquire) & RTE_DISTRIB_GET_BUF) {
				d->bufs[wid].count = 0;
				release(d, wid);
				handle_returns(d, wid);
			}
		}
		return 0;
	}

	if (unlikely(!d->activesum))
		return 0;

	while (next_idx < num_mbufs) {
		uint16_t matches[RTE_DIST_BURST_SIZE] __rte_aligned(128);
		unsigned int pkts;

		if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE)
			pkts = num_mbufs - next_idx;
		else
			pkts = RTE_DIST_BURST_SIZE;

		for (i = 0; i < pkts; i++) {
			if (mbufs[next_idx + i]) {
				/* flows have to be non-zero */
				flows[i] = mbufs[next_idx + i]->hash.usr | 1;
			} else
				flows[i] = 0;
		}
		for (; i < RTE_DIST_BURST_SIZE; i++)
			flows[i] = 0;

		matching_required = 1;