src/dpdk/lib/librte_distributor/rte_distributor_single.c
int
rte_distributor_process_single(struct rte_distributor_single *d,
		struct rte_mbuf **mbufs, unsigned num_mbufs)
{
	unsigned next_idx = 0;
	unsigned wkr = 0;
	struct rte_mbuf *next_mb = NULL;
	int64_t next_value = 0;
	uint32_t new_tag = 0;
	unsigned ret_start = d->returns.start,
			ret_count = d->returns.count;

	if (unlikely(num_mbufs == 0))
		return process_returns(d);

	while (next_idx < num_mbufs || next_mb != NULL) {
		uintptr_t oldbuf = 0;
		/* Sync with worker. Acquire bufptr64. */
		int64_t data = __atomic_load_n(&(d->bufs[wkr].bufptr64),
						__ATOMIC_ACQUIRE);

		if (!next_mb) {
			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.
			 */
			new_tag = next_mb->hash.usr;

			/*
			 * Note that if RTE_DISTRIB_MAX_WORKERS is larger than 64
			 * then the size of match has to be expanded.
			 */
			uint64_t match = 0;
			unsigned i;
			/*
			 * to scan for a match use "xor" and "not" to get a 0/1
			 * value, then use shifting to merge to single "match"
			 * variable, where a one-bit indicates a match for the
			 * worker given by the bit-position
			 */
			for (i = 0; i < d->num_workers; i++)
				match |= (!(d->in_flight_tags[i] ^ new_tag)
					<< i);

			/* Only turned-on bits are considered as match */
			match &= d->in_flight_bitmask;

			if (match) {
				next_mb = NULL;
				unsigned worker = __builtin_ctzl(match);
				if (add_to_backlog(&d->backlog[worker],
						next_value) < 0)
					next_idx--;
			}
		}

		if ((data & RTE_DISTRIB_GET_BUF) &&
				(d->backlog[wkr].count || next_mb)) {

			if (d->backlog[wkr].count)
				/* Sync with worker. Release bufptr64. */
				__atomic_store_n(&(d->bufs[wkr].bufptr64),
						backlog_pop(&d->backlog[wkr]),
						__ATOMIC_RELEASE);

			else {
				/* Sync with worker. Release bufptr64.  */
				__atomic_store_n(&(d->bufs[wkr].bufptr64),
						next_value,
						__ATOMIC_RELEASE);
				d->in_flight_tags[wkr] = new_tag;
				d->in_flight_bitmask |= (1UL << wkr);
				next_mb = NULL;
			}
			oldbuf = data >> RTE_DISTRIB_FLAG_BITS;
		} else if (data & RTE_DISTRIB_RETURN_BUF) {
			handle_worker_shutdown(d, wkr);
			oldbuf = data >> RTE_DISTRIB_FLAG_BITS;
		}

		/* store returns in a circular buffer */
		store_return(oldbuf, d, &ret_start, &ret_count);

		if (++wkr == d->num_workers)
			wkr = 0;
	}
	/* to finish, check all workers for backlog and schedule work for them
	 * if they are ready */
	for (wkr = 0; wkr < d->num_workers; wkr++)
		if (d->backlog[wkr].count &&
				/* Sync with worker. Acquire bufptr64. */
				(__atomic_load_n(&(d->bufs[wkr].bufptr64),
				__ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)) {

			int64_t oldbuf = d->bufs[wkr].bufptr64 >>
					RTE_DISTRIB_FLAG_BITS;

			store_return(oldbuf, d, &ret_start, &ret_count);

			/* Sync with worker. Release bufptr64. */
			__atomic_store_n(&(d->bufs[wkr].bufptr64),
				backlog_pop(&d->backlog[wkr]),
				__ATOMIC_RELEASE);
		}

	d->returns.start = ret_start;
	d->returns.count = ret_count;
	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_single(struct rte_distributor_single *d,
		struct rte_mbuf **mbufs, unsigned num_mbufs)
{
	unsigned next_idx = 0;
	unsigned wkr = 0;
	struct rte_mbuf *next_mb = NULL;
	int64_t next_value = 0;
	uint32_t new_tag = 0;
	unsigned ret_start = d->returns.start,
			ret_count = d->returns.count;
	if (unlikely(num_mbufs == 0))
		return process_returns(d);
	while (next_idx < num_mbufs || next_mb != NULL) {
		uintptr_t oldbuf = 0;
		int64_t data = __atomic_load_n(&(d->bufs[wkr].bufptr64),
						__ATOMIC_ACQUIRE);
		if (!next_mb) {
			next_mb = mbufs[next_idx++];
			next_value = (((int64_t)(uintptr_t)next_mb)
					<< RTE_DISTRIB_FLAG_BITS);
			new_tag = next_mb->hash.usr;
			uint64_t match = 0;
			unsigned i;
			for (i = 0; i < d->num_workers; i++)
				match |= (!(d->in_flight_tags[i] ^ new_tag)
					<< i);
			match &= d->in_flight_bitmask;
			if (match) {
				next_mb = NULL;
				unsigned worker = __builtin_ctzl(match);
				if (add_to_backlog(&d->backlog[worker],
						next_value) < 0)
					next_idx--;
			}
		}
		if ((data & RTE_DISTRIB_GET_BUF) &&
				(d->backlog[wkr].count || next_mb)) {
			if (d->backlog[wkr].count)
				__atomic_store_n(&(d->bufs[wkr].bufptr64),
						backlog_pop(&d->backlog[wkr]),
						__ATOMIC_RELEASE);
			else {
				__atomic_store_n(&(d->bufs[wkr].bufptr64),
						next_value,
						__ATOMIC_RELEASE);
				d->in_flight_tags[wkr] = new_tag;
				d->in_flight_bitmask |= (1UL << wkr);
				next_mb = NULL;
			}
			oldbuf = data >> RTE_DISTRIB_FLAG_BITS;
		} else if (data & RTE_DISTRIB_RETURN_BUF) {
			handle_worker_shutdown(d, wkr);
			oldbuf = data >> RTE_DISTRIB_FLAG_BITS;
		}
		store_return(oldbuf, d, &ret_start, &ret_count);
		if (++wkr == d->num_workers)
			wkr = 0;
	}
	for (wkr = 0; wkr < d->num_workers; wkr++)
		if (d->backlog[wkr].count &&
				(__atomic_load_n(&(d->bufs[wkr].bufptr64),
				__ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)) {
			int64_t oldbuf = d->bufs[wkr].bufptr64 >>
					RTE_DISTRIB_FLAG_BITS;
			store_return(oldbuf, d, &ret_start, &ret_count);
			__atomic_store_n(&(d->bufs[wkr].bufptr64),
				backlog_pop(&d->backlog[wkr]),
				__ATOMIC_RELEASE);
		}
	d->returns.start = ret_start;
	d->returns.count = ret_count;
	return num_mbufs;
}
int rte_distributor_process_single ( struct rte_distributor_single * d , struct rte_mbuf * * mbufs , unsigned num_mbufs ) { unsigned next_idx = 0 ; unsigned wkr = 0 ; struct rte_mbuf * next_mb = NULL ; int64_t next_value = 0 ; uint32_t new_tag = 0 ; unsigned ret_start = d -> returns . start , ret_count = d -> returns . count ; if ( unlikely ( num_mbufs == 0 ) ) return process_returns ( d ) ; while ( next_idx < num_mbufs || next_mb != NULL ) { uintptr_t oldbuf = 0 ; int64_t data = __atomic_load_n ( & ( d -> bufs [ wkr ] . bufptr64 ) , __ATOMIC_ACQUIRE ) ; if ( ! next_mb ) { next_mb = mbufs [ next_idx ++ ] ; next_value = ( ( ( int64_t ) ( uintptr_t ) next_mb ) << RTE_DISTRIB_FLAG_BITS ) ; new_tag = next_mb -> hash . usr ; uint64_t match = 0 ; unsigned i ; for ( i = 0 ; i < d -> num_workers ; i ++ ) match |= ( ! ( d -> in_flight_tags [ i ] ^ new_tag ) << i ) ; match &= d -> in_flight_bitmask ; if ( match ) { next_mb = NULL ; unsigned worker = __builtin_ctzl ( match ) ; if ( add_to_backlog ( & d -> backlog [ worker ] , next_value ) < 0 ) next_idx -- ; } } if ( ( data & RTE_DISTRIB_GET_BUF ) && ( d -> backlog [ wkr ] . count || next_mb ) ) { if ( d -> backlog [ wkr ] . count ) __atomic_store_n ( & ( d -> bufs [ wkr ] . bufptr64 ) , backlog_pop ( & d -> backlog [ wkr ] ) , __ATOMIC_RELEASE ) ; else { __atomic_store_n ( & ( d -> bufs [ wkr ] . bufptr64 ) , next_value , __ATOMIC_RELEASE ) ; d -> in_flight_tags [ wkr ] = new_tag ; d -> in_flight_bitmask |= ( 1UL << wkr ) ; next_mb = NULL ; } oldbuf = data >> RTE_DISTRIB_FLAG_BITS ; } else if ( data & RTE_DISTRIB_RETURN_BUF ) { handle_worker_shutdown ( d , wkr ) ; oldbuf = data >> RTE_DISTRIB_FLAG_BITS ; } store_return ( oldbuf , d , & ret_start , & ret_count ) ; if ( ++ wkr == d -> num_workers ) wkr = 0 ; } for ( wkr = 0 ; wkr < d -> num_workers ; wkr ++ ) if ( d -> backlog [ wkr ] . count && ( __atomic_load_n ( & ( d -> bufs [ wkr ] . bufptr64 ) , __ATOMIC_ACQUIRE ) & RTE_DISTRIB_GET_BUF ) ) { int64_t oldbuf = d -> bufs [ wkr ] . bufptr64 >> RTE_DISTRIB_FLAG_BITS ; store_return ( oldbuf , d , & ret_start , & ret_count ) ; __atomic_store_n ( & ( d -> bufs [ wkr ] . bufptr64 ) , backlog_pop ( & d -> backlog [ wkr ] ) , __ATOMIC_RELEASE ) ; } d -> returns . start = ret_start ; d -> returns . count = ret_count ; return num_mbufs ; }
process a set of packets to distribute them to workers
process a set of packets to distribute them to workers
/* Sync with worker. Acquire bufptr64. */ /*
			 * 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.
			 */ /*
			 * Note that if RTE_DISTRIB_MAX_WORKERS is larger than 64
			 * then the size of match has to be expanded.
			 */ /*
			 * to scan for a match use "xor" and "not" to get a 0/1
			 * value, then use shifting to merge to single "match"
			 * variable, where a one-bit indicates a match for the
			 * worker given by the bit-position
			 */ /* Only turned-on bits are considered as match */ /* Sync with worker. Release bufptr64. */ /* Sync with worker. Release bufptr64.  */ /* store returns in a circular buffer */ /* to finish, check all workers for backlog and schedule work for them
	 * if they are ready */ /* Sync with worker. Acquire bufptr64. */ /* Sync with worker. Release bufptr64. */