src/dpdk/lib/librte_distributor/rte_distributor.c
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 (__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
				__ATOMIC_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];
		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;

		for (j = 0; j < pkts; j++) {
			if (unlikely(!d->activesum))
				return next_idx;

			if (unlikely(matching_required)) {
				switch (d->dist_match_fn) {
				case RTE_DIST_MATCH_VECTOR:
					find_match_vec(d, &flows[0],
						&matches[0]);
					break;
				default:
					find_match_scalar(d, &flows[0],
						&matches[0]);
				}
				matching_required = 0;
			}
		/*
		 * Matches array now contain the intended worker ID (+1) of
		 * the incoming packets. Any zeroes need to be assigned
		 * workers.
		 */

			next_mb = mbufs[next_idx++];
			next_value = (((int64_t)(uintptr_t)next_mb) <<
					RTE_DISTRIB_FLAG_BITS);
			/*
			 * User is advocated to set tag value for each
			 * mbuf before calling rte_distributor_process.
			 * User defined tags are used to identify flows,
			 * or sessions.
			 */
			/* flows MUST be non-zero */
			new_tag = (uint16_t)(next_mb->hash.usr) | 1;

			/*
			 * Uncommenting the next line will cause the find_match
			 * function to be optimized out, making this function
			 * do parallel (non-atomic) distribution
			 */
			/* matches[j] = 0; */

			if (matches[j] && d->active[matches[j]-1]) {
				struct rte_distributor_backlog *bl =
						&d->backlog[matches[j]-1];
				if (unlikely(bl->count ==
						RTE_DIST_BURST_SIZE)) {
					release(d, matches[j]-1);
					if (!d->active[matches[j]-1]) {
						j--;
						next_idx--;
						matching_required = 1;
						continue;
					}
				}

				/* Add to worker that already has flow */
				unsigned int idx = bl->count++;

				bl->tags[idx] = new_tag;
				bl->pkts[idx] = next_value;

			} else {
				struct rte_distributor_backlog *bl;

				while (unlikely(!d->active[wkr]))
					wkr = (wkr + 1) % d->num_workers;
				bl = &d->backlog[wkr];

				if (unlikely(bl->count ==
						RTE_DIST_BURST_SIZE)) {
					release(d, wkr);
					if (!d->active[wkr]) {
						j--;
						next_idx--;
						matching_required = 1;
						continue;
					}
				}

				/* Add to current worker worker */
				unsigned int idx = bl->count++;

				bl->tags[idx] = new_tag;
				bl->pkts[idx] = next_value;
				/*
				 * Now that we've just added an unpinned flow
				 * to a worker, we need to ensure that all
				 * other packets with that same flow will go
				 * to the same worker in this burst.
				 */
				for (w = j; w < pkts; w++)
					if (flows[w] == new_tag)
						matches[w] = wkr+1;
			}
		}
		wkr = (wkr + 1) % d->num_workers;
	}

	/* 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 ((__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
			__ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)) {
			d->bufs[wid].count = 0;
			release(d, wid);
		}

	return num_mbufs;
}
/* process a set of packets to distribute them to workers */
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) {
		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)) {
		for (wid = 0 ; wid < d->num_workers; wid++) {
			if (__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
				__ATOMIC_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];
		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[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;
		for (j = 0; j < pkts; j++) {
			if (unlikely(!d->activesum))
				return next_idx;
			if (unlikely(matching_required)) {
				switch (d->dist_match_fn) {
				case RTE_DIST_MATCH_VECTOR:
					find_match_vec(d, &flows[0],
						&matches[0]);
					break;
				default:
					find_match_scalar(d, &flows[0],
						&matches[0]);
				}
				matching_required = 0;
			}
			next_mb = mbufs[next_idx++];
			next_value = (((int64_t)(uintptr_t)next_mb) <<
					RTE_DISTRIB_FLAG_BITS);
			new_tag = (uint16_t)(next_mb->hash.usr) | 1;
			if (matches[j] && d->active[matches[j]-1]) {
				struct rte_distributor_backlog *bl =
						&d->backlog[matches[j]-1];
				if (unlikely(bl->count ==
						RTE_DIST_BURST_SIZE)) {
					release(d, matches[j]-1);
					if (!d->active[matches[j]-1]) {
						j--;
						next_idx--;
						matching_required = 1;
						continue;
					}
				}
				unsigned int idx = bl->count++;
				bl->tags[idx] = new_tag;
				bl->pkts[idx] = next_value;
			} else {
				struct rte_distributor_backlog *bl;
				while (unlikely(!d->active[wkr]))
					wkr = (wkr + 1) % d->num_workers;
				bl = &d->backlog[wkr];
				if (unlikely(bl->count ==
						RTE_DIST_BURST_SIZE)) {
					release(d, wkr);
					if (!d->active[wkr]) {
						j--;
						next_idx--;
						matching_required = 1;
						continue;
					}
				}
				unsigned int idx = bl->count++;
				bl->tags[idx] = new_tag;
				bl->pkts[idx] = next_value;
				for (w = j; w < pkts; w++)
					if (flows[w] == new_tag)
						matches[w] = wkr+1;
			}
		}
		wkr = (wkr + 1) % d->num_workers;
	}
	for (wid = 0 ; wid < d->num_workers; wid++)
		if ((__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
			__ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)) {
			d->bufs[wid].count = 0;
			release(d, wid);
		}
	return num_mbufs;
}
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 ) { 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 ) ) { for ( wid = 0 ; wid < d -> num_workers ; wid ++ ) { if ( __atomic_load_n ( & ( d -> bufs [ wid ] . bufptr64 [ 0 ] ) , __ATOMIC_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 ] ; 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 [ 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 ; for ( j = 0 ; j < pkts ; j ++ ) { if ( unlikely ( ! d -> activesum ) ) return next_idx ; if ( unlikely ( matching_required ) ) { switch ( d -> dist_match_fn ) { case RTE_DIST_MATCH_VECTOR : find_match_vec ( d , & flows [ 0 ] , & matches [ 0 ] ) ; break ; default : find_match_scalar ( d , & flows [ 0 ] , & matches [ 0 ] ) ; } matching_required = 0 ; } next_mb = mbufs [ next_idx ++ ] ; next_value = ( ( ( int64_t ) ( uintptr_t ) next_mb ) << RTE_DISTRIB_FLAG_BITS ) ; new_tag = ( uint16_t ) ( next_mb -> hash . usr ) | 1 ; if ( matches [ j ] && d -> active [ matches [ j ] - 1 ] ) { struct rte_distributor_backlog * bl = & d -> backlog [ matches [ j ] - 1 ] ; if ( unlikely ( bl -> count == RTE_DIST_BURST_SIZE ) ) { release ( d , matches [ j ] - 1 ) ; if ( ! d -> active [ matches [ j ] - 1 ] ) { j -- ; next_idx -- ; matching_required = 1 ; continue ; } } unsigned int idx = bl -> count ++ ; bl -> tags [ idx ] = new_tag ; bl -> pkts [ idx ] = next_value ; } else { struct rte_distributor_backlog * bl ; while ( unlikely ( ! d -> active [ wkr ] ) ) wkr = ( wkr + 1 ) % d -> num_workers ; bl = & d -> backlog [ wkr ] ; if ( unlikely ( bl -> count == RTE_DIST_BURST_SIZE ) ) { release ( d , wkr ) ; if ( ! d -> active [ wkr ] ) { j -- ; next_idx -- ; matching_required = 1 ; continue ; } } unsigned int idx = bl -> count ++ ; bl -> tags [ idx ] = new_tag ; bl -> pkts [ idx ] = next_value ; for ( w = j ; w < pkts ; w ++ ) if ( flows [ w ] == new_tag ) matches [ w ] = wkr + 1 ; } } wkr = ( wkr + 1 ) % d -> num_workers ; } for ( wid = 0 ; wid < d -> num_workers ; wid ++ ) if ( ( __atomic_load_n ( & ( d -> bufs [ wid ] . bufptr64 [ 0 ] ) , __ATOMIC_ACQUIRE ) & RTE_DISTRIB_GET_BUF ) ) { d -> bufs [ wid ] . count = 0 ; release ( d , wid ) ; } return num_mbufs ; }
process a set of packets to distribute them to workers
process a set of packets to distribute them to workers
/* Call the old API */ /* Flush out all non-full cache-lines to workers. */ /* Sync with worker on GET_BUF flag. */ /* flows have to be non-zero */ /*
		 * Matches array now contain the intended worker ID (+1) of
		 * the incoming packets. Any zeroes need to be assigned
		 * workers.
		 */ /*
			 * User is advocated to set tag value for each
			 * mbuf before calling rte_distributor_process.
			 * User defined tags are used to identify flows,
			 * or sessions.
			 */ /* flows MUST be non-zero */ /*
			 * Uncommenting the next line will cause the find_match
			 * function to be optimized out, making this function
			 * do parallel (non-atomic) distribution
			 */ /* matches[j] = 0; */ /* Add to worker that already has flow */ /* Add to current worker worker */ /*
				 * Now that we've just added an unpinned flow
				 * to a worker, we need to ensure that all
				 * other packets with that same flow will go
				 * to the same worker in this burst.
				 */ /* Flush out all non-full cache-lines to workers. */ /* Sync with worker on GET_BUF flag. */