Function sanity_test_with_worker_shutdown

dpdk/app/test/test_distributor.c:441–520

Perform a sanity test of the distributor with a large number of packets, * where we allocate a new set of mbufs for each burst. The workers then * free the mbufs. This ensures that we don't have any packet leaks in the * library. */

 * library.
 */
static int
sanity_test_with_worker_shutdown(struct worker_params *wp,
		struct rte_mempool *p)
{
	struct rte_distributor *d = wp->dist;
	struct rte_mbuf *bufs[BURST];
	struct rte_mbuf *bufs2[BURST];
	unsigned int i;
	unsigned int failed = 0;
	unsigned int processed = 0;

	printf("=== Sanity test of worker shutdown ===\n");

	clear_packet_count();

	if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) {
		printf("line %d: Error getting mbufs from pool\n", __LINE__);
		return -1;
	}

	/*
	 * Now set all hash values in all buffers to same value so all
	 * pkts go to the one worker thread
	 */
	for (i = 0; i < BURST; i++)
		bufs[i]->hash.usr = 1;

	processed = 0;
	while (processed < BURST)
		processed += rte_distributor_process(d, &bufs[processed],
			BURST - processed);
	rte_distributor_flush(d);

	/* at this point, we will have processed some packets and have a full
	 * backlog for the other ones at worker 0.
	 */

	/* get more buffers to queue up, again setting them to the same flow */
	if (rte_mempool_get_bulk(p, (void *)bufs2, BURST) != 0) {
		printf("line %d: Error getting mbufs from pool\n", __LINE__);
		rte_mempool_put_bulk(p, (void *)bufs, BURST);
		return -1;
	}
	for (i = 0; i < BURST; i++)
		bufs2[i]->hash.usr = 1;

	/* get worker zero to quit */
	zero_quit = 1;
	rte_distributor_process(d, bufs2, BURST);

	/* flush the distributor */
	rte_distributor_flush(d);
	while (!__atomic_load_n(&zero_sleep, __ATOMIC_ACQUIRE))
		rte_distributor_flush(d);

	zero_quit = 0;
	while (__atomic_load_n(&zero_sleep, __ATOMIC_ACQUIRE))
		rte_delay_us(100);