[ceph.git] / ceph / src / seastar / dpdk / examples / quota_watermark / qw / main.c

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright(c) 2010-2014 Intel Corporation
 */

#include <rte_eal.h>

#include <rte_common.h>
#include <rte_debug.h>
#include <rte_errno.h>
#include <rte_ethdev.h>
#include <rte_launch.h>
#include <rte_lcore.h>
#include <rte_log.h>
#include <rte_mbuf.h>
#include <rte_ring.h>

#include <rte_byteorder.h>

#include "args.h"
#include "main.h"
#include "init.h"
#include "../include/conf.h"


#ifdef QW_SOFTWARE_FC
#define SEND_PAUSE_FRAME(port_id, duration) send_pause_frame(port_id, duration)
#else
#define SEND_PAUSE_FRAME(port_id, duration) do { } while(0)
#endif

#define ETHER_TYPE_FLOW_CONTROL 0x8808

struct ether_fc_frame {
	uint16_t opcode;
	uint16_t param;
} __attribute__((__packed__));


int *quota;
unsigned int *low_watermark;
unsigned int *high_watermark;

uint16_t port_pairs[RTE_MAX_ETHPORTS];

struct rte_ring *rings[RTE_MAX_LCORE][RTE_MAX_ETHPORTS];
struct rte_mempool *mbuf_pool;


static void send_pause_frame(uint16_t port_id, uint16_t duration)
{
	struct rte_mbuf *mbuf;
	struct ether_fc_frame *pause_frame;
	struct ether_hdr *hdr;
	struct ether_addr mac_addr;

	RTE_LOG_DP(DEBUG, USER1,
			"Sending PAUSE frame (duration=%d) on port %d\n",
			duration, port_id);

	/* Get a mbuf from the pool */
	mbuf = rte_pktmbuf_alloc(mbuf_pool);
	if (unlikely(mbuf == NULL))
		return;

	/* Prepare a PAUSE frame */
	hdr = rte_pktmbuf_mtod(mbuf, struct ether_hdr *);
	pause_frame = (struct ether_fc_frame *) &hdr[1];

	rte_eth_macaddr_get(port_id, &mac_addr);
	ether_addr_copy(&mac_addr, &hdr->s_addr);

	void *tmp = &hdr->d_addr.addr_bytes[0];
	*((uint64_t *)tmp) = 0x010000C28001ULL;

	hdr->ether_type = rte_cpu_to_be_16(ETHER_TYPE_FLOW_CONTROL);

	pause_frame->opcode = rte_cpu_to_be_16(0x0001);
	pause_frame->param  = rte_cpu_to_be_16(duration);

	mbuf->pkt_len  = 60;
	mbuf->data_len = 60;

	rte_eth_tx_burst(port_id, 0, &mbuf, 1);
}

/**
 * Get the previous enabled lcore ID
 *
 * @param lcore_id
 *   The current lcore ID.
 * @return
 *   The previous enabled lcore_id or -1 if not found.
 */
static unsigned int
get_previous_lcore_id(unsigned int lcore_id)
{
	int i;

	for (i = lcore_id - 1; i >= 0; i--)
		if (rte_lcore_is_enabled(i))
			return i;

	return -1;
}

/**
 * Get the last enabled lcore ID
 *
 * @return
 *   The last enabled lcore_id.
 */
static unsigned int
get_last_lcore_id(void)
{
	int i;

	for (i = RTE_MAX_LCORE; i >= 0; i--)
		if (rte_lcore_is_enabled(i))
			return i;

	return 0;
}

static void
receive_stage(__attribute__((unused)) void *args)
{
	int i, ret;

	uint16_t port_id;
	uint16_t nb_rx_pkts;

	unsigned int lcore_id;
	unsigned int free;

	struct rte_mbuf *pkts[MAX_PKT_QUOTA];
	struct rte_ring *ring;
	enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };

	lcore_id = rte_lcore_id();

	RTE_LOG(INFO, USER1,
			"%s() started on core %u\n", __func__, lcore_id);

	while (1) {

		/* Process each port round robin style */
		for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {

			if (!is_bit_set(port_id, portmask))
				continue;

			ring = rings[lcore_id][port_id];

			if (ring_state[port_id] != RING_READY) {
				if (rte_ring_count(ring) > *low_watermark)
					continue;
				else
					ring_state[port_id] = RING_READY;
			}

			/* Enqueue received packets on the RX ring */
			nb_rx_pkts = rte_eth_rx_burst(port_id, 0, pkts,
					(uint16_t) *quota);
			ret = rte_ring_enqueue_bulk(ring, (void *) pkts,
					nb_rx_pkts, &free);
			if (RING_SIZE - free > *high_watermark) {
				ring_state[port_id] = RING_OVERLOADED;
				send_pause_frame(port_id, 1337);
			}

			if (ret == 0) {

				/*
				 * Return  mbufs to the pool,
				 * effectively dropping packets
				 */
				for (i = 0; i < nb_rx_pkts; i++)
					rte_pktmbuf_free(pkts[i]);
			}
		}
	}
}

static int
pipeline_stage(__attribute__((unused)) void *args)
{
	int i, ret;
	int nb_dq_pkts;

	uint16_t port_id;

	unsigned int lcore_id, previous_lcore_id;
	unsigned int free;

	void *pkts[MAX_PKT_QUOTA];
	struct rte_ring *rx, *tx;
	enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };

	lcore_id = rte_lcore_id();
	previous_lcore_id = get_previous_lcore_id(lcore_id);

	RTE_LOG(INFO, USER1,
			"%s() started on core %u - processing packets from core %u\n",
			__func__, lcore_id, previous_lcore_id);

	while (1) {

		for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {

			if (!is_bit_set(port_id, portmask))
				continue;

			tx = rings[lcore_id][port_id];
			rx = rings[previous_lcore_id][port_id];

			if (ring_state[port_id] != RING_READY) {
				if (rte_ring_count(tx) > *low_watermark)
					continue;
				else
					ring_state[port_id] = RING_READY;
			}

			/* Dequeue up to quota mbuf from rx */
			nb_dq_pkts = rte_ring_dequeue_burst(rx, pkts,
					*quota, NULL);
			if (unlikely(nb_dq_pkts < 0))
				continue;

			/* Enqueue them on tx */
			ret = rte_ring_enqueue_bulk(tx, pkts,
					nb_dq_pkts, &free);
			if (RING_SIZE - free > *high_watermark)
				ring_state[port_id] = RING_OVERLOADED;

			if (ret == 0) {

				/*
				 * Return  mbufs to the pool,
				 * effectively dropping packets
				 */
				for (i = 0; i < nb_dq_pkts; i++)
					rte_pktmbuf_free(pkts[i]);
			}
		}
	}

	return 0;
}

static int
send_stage(__attribute__((unused)) void *args)
{
	uint16_t nb_dq_pkts;

	uint16_t port_id;
	uint16_t dest_port_id;

	unsigned int lcore_id, previous_lcore_id;

	struct rte_ring *tx;
	struct rte_mbuf *tx_pkts[MAX_PKT_QUOTA];

	lcore_id = rte_lcore_id();
	previous_lcore_id = get_previous_lcore_id(lcore_id);

	RTE_LOG(INFO, USER1,
			"%s() started on core %u - processing packets from core %u\n",
			__func__, lcore_id, previous_lcore_id);

	while (1) {

		/* Process each ring round robin style */
		for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {

			if (!is_bit_set(port_id, portmask))
				continue;

			dest_port_id = port_pairs[port_id];
			tx = rings[previous_lcore_id][port_id];

			if (rte_ring_empty(tx))
				continue;

			/* Dequeue packets from tx and send them */
			nb_dq_pkts = (uint16_t) rte_ring_dequeue_burst(tx,
					(void *) tx_pkts, *quota, NULL);
			rte_eth_tx_burst(dest_port_id, 0, tx_pkts, nb_dq_pkts);

			/* TODO: Check if nb_dq_pkts == nb_tx_pkts? */
		}
	}

	return 0;
}

int
main(int argc, char **argv)
{
	int ret;
	unsigned int lcore_id, master_lcore_id, last_lcore_id;

	uint16_t port_id;

	rte_log_set_global_level(RTE_LOG_INFO);

	ret = rte_eal_init(argc, argv);
	if (ret < 0)
		rte_exit(EXIT_FAILURE, "Cannot initialize EAL\n");

	argc -= ret;
	argv += ret;

	init_dpdk();
	setup_shared_variables();

	*quota = 32;
	*low_watermark = 60 * RING_SIZE / 100;

	last_lcore_id   = get_last_lcore_id();
	master_lcore_id = rte_get_master_lcore();

	/* Parse the application's arguments */
	ret = parse_qw_args(argc, argv);
	if (ret < 0)
		rte_exit(EXIT_FAILURE, "Invalid quota/watermark argument(s)\n");

	/* Create a pool of mbuf to store packets */
	mbuf_pool = rte_pktmbuf_pool_create("mbuf_pool", MBUF_PER_POOL, 32, 0,
			MBUF_DATA_SIZE, rte_socket_id());
	if (mbuf_pool == NULL)
		rte_panic("%s\n", rte_strerror(rte_errno));

	for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
		if (is_bit_set(port_id, portmask)) {
			configure_eth_port(port_id);
			init_ring(master_lcore_id, port_id);
		}

	pair_ports();

	/*
	 * Start pipeline_connect() on all the available slave lcores
	 * but the last
	 */
	for (lcore_id = 0 ; lcore_id < last_lcore_id; lcore_id++) {
		if (rte_lcore_is_enabled(lcore_id) &&
				lcore_id != master_lcore_id) {

			for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
				if (is_bit_set(port_id, portmask))
					init_ring(lcore_id, port_id);

			rte_eal_remote_launch(pipeline_stage,
					NULL, lcore_id);
		}
	}

	/* Start send_stage() on the last slave core */
	rte_eal_remote_launch(send_stage, NULL, last_lcore_id);

	/* Start receive_stage() on the master core */
	receive_stage(NULL);

	return 0;
}
Commit	Line	Data
9f95a23c TL	1	/* SPDX-License-Identifier: BSD-3-Clause
9f95a23c TL	2	* Copyright(c) 2010-2014 Intel Corporation
11fdf7f2 TL	3	*/
	4
	5	#include <rte_eal.h>
	6
	7	#include <rte_common.h>
	8	#include <rte_debug.h>
	9	#include <rte_errno.h>
	10	#include <rte_ethdev.h>
	11	#include <rte_launch.h>
	12	#include <rte_lcore.h>
	13	#include <rte_log.h>
	14	#include <rte_mbuf.h>
	15	#include <rte_ring.h>
	16
	17	#include <rte_byteorder.h>
	18
	19	#include "args.h"
	20	#include "main.h"
	21	#include "init.h"
	22	#include "../include/conf.h"
	23
	24
	25	#ifdef QW_SOFTWARE_FC
	26	#define SEND_PAUSE_FRAME(port_id, duration) send_pause_frame(port_id, duration)
	27	#else
	28	#define SEND_PAUSE_FRAME(port_id, duration) do { } while(0)
	29	#endif
	30
	31	#define ETHER_TYPE_FLOW_CONTROL 0x8808
	32
	33	struct ether_fc_frame {
	34	uint16_t opcode;
	35	uint16_t param;
	36	} __attribute__((__packed__));
	37
	38
	39	int *quota;
	40	unsigned int *low_watermark;
	41	unsigned int *high_watermark;
	42
9f95a23c	43	uint16_t port_pairs[RTE_MAX_ETHPORTS];
11fdf7f2 TL	44
	45	struct rte_ring *rings[RTE_MAX_LCORE][RTE_MAX_ETHPORTS];
	46	struct rte_mempool *mbuf_pool;
	47
	48
9f95a23c	49	static void send_pause_frame(uint16_t port_id, uint16_t duration)
11fdf7f2 TL	50	{
	51	struct rte_mbuf *mbuf;
	52	struct ether_fc_frame *pause_frame;
	53	struct ether_hdr *hdr;
	54	struct ether_addr mac_addr;
	55
	56	RTE_LOG_DP(DEBUG, USER1,
	57	"Sending PAUSE frame (duration=%d) on port %d\n",
	58	duration, port_id);
	59
	60	/* Get a mbuf from the pool */
	61	mbuf = rte_pktmbuf_alloc(mbuf_pool);
	62	if (unlikely(mbuf == NULL))
	63	return;
	64
	65	/* Prepare a PAUSE frame */
	66	hdr = rte_pktmbuf_mtod(mbuf, struct ether_hdr *);
	67	pause_frame = (struct ether_fc_frame *) &hdr[1];
	68
	69	rte_eth_macaddr_get(port_id, &mac_addr);
	70	ether_addr_copy(&mac_addr, &hdr->s_addr);
	71
	72	void *tmp = &hdr->d_addr.addr_bytes[0];
	73	((uint64_t )tmp) = 0x010000C28001ULL;
	74
	75	hdr->ether_type = rte_cpu_to_be_16(ETHER_TYPE_FLOW_CONTROL);
	76
	77	pause_frame->opcode = rte_cpu_to_be_16(0x0001);
	78	pause_frame->param = rte_cpu_to_be_16(duration);
	79
	80	mbuf->pkt_len = 60;
	81	mbuf->data_len = 60;
	82
	83	rte_eth_tx_burst(port_id, 0, &mbuf, 1);
	84	}
	85
	86	/**
	87	* Get the previous enabled lcore ID
	88	*
	89	* @param lcore_id
	90	* The current lcore ID.
	91	* @return
	92	* The previous enabled lcore_id or -1 if not found.
	93	*/
	94	static unsigned int
	95	get_previous_lcore_id(unsigned int lcore_id)
	96	{
	97	int i;
	98
	99	for (i = lcore_id - 1; i >= 0; i--)
	100	if (rte_lcore_is_enabled(i))
	101	return i;
	102
	103	return -1;
	104	}
	105
	106	/**
	107	* Get the last enabled lcore ID
	108	*
	109	* @return
	110	* The last enabled lcore_id.
	111	*/
	112	static unsigned int
	113	get_last_lcore_id(void)
114	{
115	int i;
116
117	for (i = RTE_MAX_LCORE; i >= 0; i--)
118	if (rte_lcore_is_enabled(i))
119	return i;
120
121	return 0;
122	}
123
124	static void
125	receive_stage(__attribute__((unused)) void *args)
126	{
127	int i, ret;
128
9f95a23c	129	uint16_t port_id;
11fdf7f2 TL	130	uint16_t nb_rx_pkts;
	131
	132	unsigned int lcore_id;
	133	unsigned int free;
	134
	135	struct rte_mbuf *pkts[MAX_PKT_QUOTA];
	136	struct rte_ring *ring;
	137	enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };
	138
	139	lcore_id = rte_lcore_id();
	140
	141	RTE_LOG(INFO, USER1,
	142	"%s() started on core %u\n", __func__, lcore_id);
	143
	144	while (1) {
	145
	146	/* Process each port round robin style */
	147	for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
	148
	149	if (!is_bit_set(port_id, portmask))
	150	continue;
	151
	152	ring = rings[lcore_id][port_id];
	153
	154	if (ring_state[port_id] != RING_READY) {
	155	if (rte_ring_count(ring) > *low_watermark)
	156	continue;
	157	else
	158	ring_state[port_id] = RING_READY;
	159	}
	160
	161	/* Enqueue received packets on the RX ring */
	162	nb_rx_pkts = rte_eth_rx_burst(port_id, 0, pkts,
	163	(uint16_t) *quota);
	164	ret = rte_ring_enqueue_bulk(ring, (void *) pkts,
	165	nb_rx_pkts, &free);
	166	if (RING_SIZE - free > *high_watermark) {
	167	ring_state[port_id] = RING_OVERLOADED;
	168	send_pause_frame(port_id, 1337);
	169	}
	170
	171	if (ret == 0) {
	172
	173	/*
	174	* Return mbufs to the pool,
	175	* effectively dropping packets
	176	*/
	177	for (i = 0; i < nb_rx_pkts; i++)
	178	rte_pktmbuf_free(pkts[i]);
	179	}
	180	}
	181	}
	182	}
	183
9f95a23c	184	static int
11fdf7f2 TL	185	pipeline_stage(__attribute__((unused)) void *args)
	186	{
	187	int i, ret;
	188	int nb_dq_pkts;
	189
9f95a23c	190	uint16_t port_id;
11fdf7f2 TL	191
	192	unsigned int lcore_id, previous_lcore_id;
	193	unsigned int free;
	194
	195	void *pkts[MAX_PKT_QUOTA];
	196	struct rte_ring rx, tx;
	197	enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };
	198
	199	lcore_id = rte_lcore_id();
	200	previous_lcore_id = get_previous_lcore_id(lcore_id);
	201
	202	RTE_LOG(INFO, USER1,
	203	"%s() started on core %u - processing packets from core %u\n",
	204	__func__, lcore_id, previous_lcore_id);
	205
	206	while (1) {
	207
	208	for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
	209
	210	if (!is_bit_set(port_id, portmask))
	211	continue;
	212
	213	tx = rings[lcore_id][port_id];
	214	rx = rings[previous_lcore_id][port_id];
	215
	216	if (ring_state[port_id] != RING_READY) {
	217	if (rte_ring_count(tx) > *low_watermark)
	218	continue;
	219	else
	220	ring_state[port_id] = RING_READY;
	221	}
	222
	223	/* Dequeue up to quota mbuf from rx */
	224	nb_dq_pkts = rte_ring_dequeue_burst(rx, pkts,
	225	*quota, NULL);
	226	if (unlikely(nb_dq_pkts < 0))
	227	continue;
	228
	229	/* Enqueue them on tx */
	230	ret = rte_ring_enqueue_bulk(tx, pkts,
	231	nb_dq_pkts, &free);
	232	if (RING_SIZE - free > *high_watermark)
	233	ring_state[port_id] = RING_OVERLOADED;
	234
	235	if (ret == 0) {
	236
	237	/*
	238	* Return mbufs to the pool,
	239	* effectively dropping packets
	240	*/
	241	for (i = 0; i < nb_dq_pkts; i++)
	242	rte_pktmbuf_free(pkts[i]);
	243	}
	244	}
	245	}
9f95a23c TL	246
9f95a23c TL	247	return 0;
11fdf7f2 TL	248	}
11fdf7f2 TL	249
9f95a23c	250	static int
11fdf7f2 TL	251	send_stage(__attribute__((unused)) void *args)
	252	{
	253	uint16_t nb_dq_pkts;
	254
9f95a23c TL	255	uint16_t port_id;
9f95a23c TL	256	uint16_t dest_port_id;
11fdf7f2 TL	257
	258	unsigned int lcore_id, previous_lcore_id;
	259
	260	struct rte_ring *tx;
	261	struct rte_mbuf *tx_pkts[MAX_PKT_QUOTA];
	262
	263	lcore_id = rte_lcore_id();
	264	previous_lcore_id = get_previous_lcore_id(lcore_id);
	265
	266	RTE_LOG(INFO, USER1,
	267	"%s() started on core %u - processing packets from core %u\n",
	268	__func__, lcore_id, previous_lcore_id);
	269
	270	while (1) {
	271
	272	/* Process each ring round robin style */
	273	for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
	274
	275	if (!is_bit_set(port_id, portmask))
	276	continue;
	277
	278	dest_port_id = port_pairs[port_id];
	279	tx = rings[previous_lcore_id][port_id];
	280
	281	if (rte_ring_empty(tx))
	282	continue;
	283
	284	/* Dequeue packets from tx and send them */
	285	nb_dq_pkts = (uint16_t) rte_ring_dequeue_burst(tx,
	286	(void ) tx_pkts, quota, NULL);
	287	rte_eth_tx_burst(dest_port_id, 0, tx_pkts, nb_dq_pkts);
	288
	289	/* TODO: Check if nb_dq_pkts == nb_tx_pkts? */
	290	}
	291	}
9f95a23c TL	292
9f95a23c TL	293	return 0;
11fdf7f2 TL	294	}
	295
	296	int
	297	main(int argc, char **argv)
	298	{
	299	int ret;
	300	unsigned int lcore_id, master_lcore_id, last_lcore_id;
	301
9f95a23c	302	uint16_t port_id;
11fdf7f2 TL	303
	304	rte_log_set_global_level(RTE_LOG_INFO);
	305
	306	ret = rte_eal_init(argc, argv);
	307	if (ret < 0)
	308	rte_exit(EXIT_FAILURE, "Cannot initialize EAL\n");
	309
	310	argc -= ret;
	311	argv += ret;
	312
	313	init_dpdk();
	314	setup_shared_variables();
	315
	316	*quota = 32;
	317	low_watermark = 60 RING_SIZE / 100;
	318
	319	last_lcore_id = get_last_lcore_id();
	320	master_lcore_id = rte_get_master_lcore();
	321
	322	/* Parse the application's arguments */
	323	ret = parse_qw_args(argc, argv);
	324	if (ret < 0)
	325	rte_exit(EXIT_FAILURE, "Invalid quota/watermark argument(s)\n");
	326
	327	/* Create a pool of mbuf to store packets */
	328	mbuf_pool = rte_pktmbuf_pool_create("mbuf_pool", MBUF_PER_POOL, 32, 0,
	329	MBUF_DATA_SIZE, rte_socket_id());
	330	if (mbuf_pool == NULL)
	331	rte_panic("%s\n", rte_strerror(rte_errno));
	332
	333	for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
	334	if (is_bit_set(port_id, portmask)) {
	335	configure_eth_port(port_id);
	336	init_ring(master_lcore_id, port_id);
	337	}
	338
	339	pair_ports();
	340
	341	/*
	342	* Start pipeline_connect() on all the available slave lcores
	343	* but the last
	344	*/
	345	for (lcore_id = 0 ; lcore_id < last_lcore_id; lcore_id++) {
	346	if (rte_lcore_is_enabled(lcore_id) &&
	347	lcore_id != master_lcore_id) {
	348
	349	for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
	350	if (is_bit_set(port_id, portmask))
	351	init_ring(lcore_id, port_id);
	352
9f95a23c	353	rte_eal_remote_launch(pipeline_stage,
11fdf7f2 TL	354	NULL, lcore_id);
	355	}
	356	}
	357
	358	/* Start send_stage() on the last slave core */
9f95a23c	359	rte_eal_remote_launch(send_stage, NULL, last_lcore_id);
11fdf7f2 TL	360
	361	/* Start receive_stage() on the master core */
	362	receive_stage(NULL);
	363
	364	return 0;
	365	}