Update dpdk: 5Mpps TX and RX

.gitignore

@@ -8,3 +8,4 @@ af-xdp-test/af_xdp.o
 raw-socket-test/raw_socket
 raw-socket-test/raw_socket.d
 raw-socket-test/raw_socket.o
+dpdk-test/build

af-xdp-test/af_xdp.c

@@ -52,8 +52,8 @@
 
 #define BUFFER_SIZE 4096
 
-#define NB_PACKETS 1000000
-#define BATCH_SIZE 2048
+#define NB_PACKETS 2000000
+#define BATCH_SIZE 256
 
 //#define PRINT_PROGRESS
 #define PRINT_PROGRESS_INTERVAL 1000
@@ -151,7 +151,7 @@ static int batch_size = BATCH_SIZE;
 int trace_fd;
 
 int main() {
-  char * network_if = "ens9f1np1";
+  char * network_if = "ens5f0np0";
 
   log_info("", "Starting");
 

dpdk-test/Makefile

+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2010-2014 Intel Corporation
+
+# binary name
+APP = dpdk-recv-send
+
+# all source are stored in SRCS-y
+SRCS-y := $(APP).c
+
+PKGCONF ?= pkg-config
+
+# Build using pkg-config variables if possible
+ifneq ($(shell $(PKGCONF) --exists libdpdk && echo 0),0)
+$(error "no installation of DPDK found")
+endif
+
+all: shared
+.PHONY: shared static
+shared: build/$(APP)-shared
+	ln -sf $(APP)-shared build/$(APP)
+static: build/$(APP)-static
+	ln -sf $(APP)-static build/$(APP)
+
+PC_FILE := $(shell $(PKGCONF) --path libdpdk 2>/dev/null)
+CFLAGS += -O3 $(shell $(PKGCONF) --cflags libdpdk)
+LDFLAGS_SHARED = $(shell $(PKGCONF) --libs libdpdk) -lpthread
+LDFLAGS_STATIC = $(shell $(PKGCONF) --static --libs libdpdk)
+
+CFLAGS += -DALLOW_EXPERIMENTAL_API
+
+build/$(APP)-shared: $(SRCS-y) Makefile $(PC_FILE) | build
+	$(CC) $(CFLAGS) $(SRCS-y) -o $@ $(LDFLAGS) $(LDFLAGS_SHARED)
+
+build/$(APP)-static: $(SRCS-y) Makefile $(PC_FILE) | build
+	$(CC) $(CFLAGS) $(SRCS-y) -o $@ $(LDFLAGS) $(LDFLAGS_STATIC)
+
+build:
+	@mkdir -p $@
+
+.PHONY: clean
+clean:
+	rm -f build/$(APP) build/$(APP)-static build/$(APP)-shared
+	test -d build && rmdir -p build || true

dpdk-test/dpdk-recv-send.c

+#define _GNU_SOURCE             /* See feature_test_macros(7) */
+#include <arpa/inet.h>
+#include <assert.h>
+#include <errno.h>
+#include <error.h>
+#include <fcntl.h>
+#include <getopt.h>
+#include <immintrin.h>
+#include <inttypes.h>
+#include <limits.h>
+#include <linux/if_packet.h>
+#include <math.h>
+#include <netdb.h>
+#include <netinet/ether.h>
+#include <netinet/in.h>
+#include <net/if.h>
+#include <pthread.h>
+#include <sched.h>
+#include <semaphore.h>
+#include <signal.h>
+#include <stdarg.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/ioctl.h>
+#include <sys/mman.h>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <time.h>
+#include <unistd.h>
+
+#include <rte_eal.h>
+#include <rte_ethdev.h>
+#include <rte_ether.h>
+#include <rte_cycles.h>
+#include <rte_lcore.h>
+#include <rte_mbuf.h>
+#include <rte_ether.h>
+#include <rte_ip.h>
+#include <rte_udp.h>
+
+static void log_error(const char * section, const char * msg, ...) {
+  time_t t;
+  struct tm ts;
+	char line[256];
+  va_list arglist;
+
+  time(&t);
+  ts = *localtime(&t);
+  strftime(line, 80, "%m-%d %H:%M:%S", &ts);
+  sprintf(line + strlen(line), " ERROR [%s] ", section);
+  va_start(arglist, msg);
+  vsprintf(line + strlen(line), msg, arglist);
+  va_end(arglist);
+  puts(line);
+  exit(EXIT_FAILURE);
+}
+static void log_info(const char * section, const char * msg, ...) {
+  time_t t;
+  struct tm ts;
+	char line[256];
+  va_list arglist;
+
+  time(&t);
+  ts = *localtime(&t);
+  strftime(line, 80, "%m-%d %H:%M:%S", &ts);
+  sprintf(line + strlen(line), " INFO [%s] ", section);
+  va_start(arglist, msg);
+  vsprintf(line + strlen(line), msg, arglist);
+  va_end(arglist);
+  puts(line);
+}
+// Timestamps utils
+#define NSEC_PER_SEC INT64_C(1000000000)
+static struct timespec int_to_ts(int64_t t) {
+  struct timespec ts;
+  ts.tv_sec = t / NSEC_PER_SEC;
+  ts.tv_nsec = t - (ts.tv_sec * NSEC_PER_SEC);
+  return ts;
+}
+static int64_t ts_to_int(struct timespec ts) {
+    return ts.tv_sec * NSEC_PER_SEC + ts.tv_nsec;
+}
+static void add_ns(struct timespec *t, int64_t ns) {
+  t->tv_nsec += ns;
+  while (t->tv_nsec >= ((int64_t)NSEC_PER_SEC)) {
+    t->tv_sec += 1;
+    t->tv_nsec -= NSEC_PER_SEC;
+  }
+}
+static int64_t calcdiff_ns(struct timespec t1, struct timespec t2) {
+  int64_t diff;
+  diff = NSEC_PER_SEC * ((int)t1.tv_sec - (int)t2.tv_sec);
+  diff += ((int)t1.tv_nsec - (int)t2.tv_nsec);
+  return diff;
+}
+
+#define BURST_SIZE 16
+//#define NB_PACKETS (1 * 1000000)
+#define NB_PACKETS 64
+#define DATA_LEN 244
+#define MEMPOOL_CACHE_SIZE 256
+#define RTE_TEST_RX_DESC_DEFAULT 1024
+#define RTE_TEST_TX_DESC_DEFAULT 1024
+static uint16_t nb_rxd = RTE_TEST_RX_DESC_DEFAULT;
+static uint16_t nb_txd = RTE_TEST_TX_DESC_DEFAULT;
+struct rte_mempool *mbuf_pool;
+struct rte_ether_addr s_addr = {{0xb8,0xce,0xf6,0x4b,0x00,0x22}};
+struct rte_ether_addr d_addr = {{0xb8,0xce,0xf6,0x4b,0x00,0x23}};
+uint8_t data[BURST_SIZE][DATA_LEN];
+
+static const struct rte_eth_conf port_conf_default = {
+	.rxmode = { .max_rx_pkt_len = RTE_ETHER_MAX_LEN }
+};
+static inline int port_init(int portid, struct rte_mempool *mbuf_pool) {
+	struct rte_eth_conf port_conf = port_conf_default;
+	const uint16_t rx_rings = 1, tx_rings = 1;
+	int retval;
+	uint16_t q;
+  int ret;
+
+	retval = rte_eth_dev_configure(portid, rx_rings, tx_rings, &port_conf);
+	if (retval != 0)
+		return retval;
+
+  //ret = rte_eth_dev_adjust_nb_rx_tx_desc(1, &nb_rxd, &nb_txd);
+  //if (ret < 0)
+  //  rte_exit(EXIT_FAILURE, "Cannot adjust number of descriptors: err=%d, port=%u\n", ret, 1);
+
+	/* Allocate and set up 1 RX queue per Ethernet port. */
+	for (q = 0; q < rx_rings; q++) {
+		retval = rte_eth_rx_queue_setup(portid, q, nb_rxd,
+				rte_eth_dev_socket_id(portid), NULL, mbuf_pool);
+		if (retval < 0)
+			return retval;
+	}
+
+	/* Allocate and set up 1 TX queue per Ethernet port. */
+	for (q = 0; q < tx_rings; q++) {
+		retval = rte_eth_tx_queue_setup(portid, q, nb_txd,
+				rte_eth_dev_socket_id(portid), NULL);
+		if (retval < 0)
+			return retval;
+	}
+
+	/* Start the Ethernet port. */
+	retval = rte_eth_dev_start(portid);
+	if (retval < 0)
+		return retval;
+
+	return 0;
+}
+static void init_dpdk(int argc, char ** argv) {
+  unsigned int nb_mbufs;
+  int ret;
+
+	ret = rte_eal_init(argc, argv);
+	if (ret < 0)
+		rte_exit(EXIT_FAILURE, "initlize fail!");
+
+	argc -= ret;
+	argv += ret;
+
+  nb_mbufs = RTE_MAX((nb_rxd + nb_txd + BURST_SIZE + MEMPOOL_CACHE_SIZE), 8192U);
+
+	mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL", nb_mbufs,
+		MEMPOOL_CACHE_SIZE, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());
+	if (mbuf_pool == NULL)
+		rte_exit(EXIT_FAILURE, "Cannot create mbuf pool\n");
+
+	if (port_init(0, mbuf_pool) != 0)
+		rte_exit(EXIT_FAILURE, "Cannot init port %"PRIu8 "\n", 0);
+	if (port_init(1, mbuf_pool) != 0)
+		rte_exit(EXIT_FAILURE, "Cannot init port %"PRIu8 "\n", 1);
+}
+
+static void send_packets(int port) {
+  struct rte_mbuf * pkt[BURST_SIZE];
+  struct rte_ether_hdr *eth_hdr;
+
+  for(int i = 0; i < BURST_SIZE; i++) {
+    int pkt_size;
+    pkt[i] = rte_pktmbuf_alloc(mbuf_pool);
+    eth_hdr = rte_pktmbuf_mtod(pkt[i], struct rte_ether_hdr*);
+    if(port) {
+      eth_hdr->d_addr = s_addr;
+      eth_hdr->s_addr = d_addr;
+    } else {
+      eth_hdr->d_addr = d_addr;
+      eth_hdr->s_addr = s_addr;
+    }
+    eth_hdr->ether_type = 0x0a00;
+    memcpy(rte_pktmbuf_mtod_offset(pkt[i], uint8_t *, sizeof(struct rte_ether_hdr)), data[i], DATA_LEN);
+    pkt_size = DATA_LEN + sizeof(struct rte_ether_hdr);
+    pkt[i]->data_len = pkt_size;
+    pkt[i]->pkt_len = pkt_size;
+  }
+  const uint16_t nb_tx = rte_eth_tx_burst(port, 0, pkt, BURST_SIZE);
+
+  /* Free any unsent packets. */
+  if (unlikely(nb_tx < BURST_SIZE)) {
+    uint16_t buf;
+    for (buf = nb_tx; buf < BURST_SIZE; buf++)
+      rte_pktmbuf_free(pkt[buf]);
+		fprintf(stderr, "Sent %d packets instead of %d\n", nb_tx, BURST_SIZE);
+		exit(EXIT_FAILURE);
+  }
+}
+
+static int recv_packets(int port) {
+  struct rte_mbuf * pkt[BURST_SIZE];
+  //if(port == 0) log_info("PORT0", "Init");
+	while(1) {
+    //if(port == 0) log_info("PORT0", "Burst begin");
+		const int nb_rx = rte_eth_rx_burst(port, 0, pkt, BURST_SIZE);
+    //if(port == 0) log_info("PORT0", "Burst end");
+
+    for(int i = 0; i < nb_rx; i++)
+      rte_pktmbuf_free(pkt[i]);
+
+		if(nb_rx)
+      return nb_rx;
+	}
+}
+
+#define FRAME_FREQ INT64_C(3840000)
+#define BURST_N ((FRAME_FREQ / 1000) / BURST_SIZE)
+#define PROGRESS_RATE 1000000
+
+void * send_thread(void * p) {
+  int * port = (int *) p;
+  int64_t seq_id = 0;
+  struct timespec next, initial, current;
+	cpu_set_t mask;
+	// Set thread CPU affinity
+	CPU_ZERO(&mask);
+	CPU_SET(37 + *port, &mask);
+	if (sched_setaffinity(0, sizeof(mask), &mask))
+		error(EXIT_FAILURE, errno, "Could not set CPU affinity to CPU %d\n", 38);
+
+  log_info("SEND_THREAD", "Init %d", *port);
+
+  for(int i = 0; i < BURST_SIZE; i++) {
+    for(int j = 0; j < DATA_LEN; j++) {
+      data[i][j] = 0xfe;
+    }
+  }
+
+	int64_t progress = PROGRESS_RATE;
+  clock_gettime(CLOCK_MONOTONIC, &initial);
+  next = initial;
+  for(int64_t sent = 0;; sent += (BURST_SIZE * BURST_N)) {
+    // Multiply by i everytime to prevent any frequence drift
+    add_ns(&next, 1000000);
+
+		if(sent > progress) {
+      clock_gettime(CLOCK_MONOTONIC, &current);
+      int64_t duration_ns = calcdiff_ns(current, initial);
+			progress += PROGRESS_RATE;
+			printf("SEND%d %" PRIi64 " (pps: %" PRIi64 ")\n", *port, sent, (sent * NSEC_PER_SEC) / duration_ns);
+		}
+
+    for(int64_t k = 0; k < BURST_N ; k++) {
+      for(int j = 0; j < BURST_SIZE; j++)
+        *(((int64_t *) data[j]) + 2) = seq_id++;
+      send_packets(*port);
+    }
+    clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &next, NULL);
+  }
+  pthread_exit(EXIT_SUCCESS);
+}
+
+void * recv_thread(void * p) {
+  int * port = (int *) p;
+	int64_t progress = PROGRESS_RATE;
+	cpu_set_t mask;
+	// Set thread CPU affinity
+	CPU_ZERO(&mask);
+	CPU_SET(35 + *port, &mask);
+	if (sched_setaffinity(0, sizeof(mask), &mask))
+		error(EXIT_FAILURE, errno, "Could not set CPU affinity to CPU %d\n", 37);
+
+  log_info("RECV_THREAD", "Init %d", *port);
+
+  for(int64_t received = 0;;) {
+		if(received > progress) {
+			progress += PROGRESS_RATE;
+			printf("RECV%d %" PRIi64 "\n", *port, received);
+		}
+		
+		received += recv_packets(*port);
+  }
+  pthread_exit(EXIT_SUCCESS);
+}
+
+int port0 = 0;
+int port1 = 1;
+
+int main(int argc, char **argv)
+{
+
+  pthread_t recv_pthread;
+  pthread_t send_pthread;
+
+  struct sched_param recv_param;
+  struct sched_param send_param;
+  pthread_attr_t recv_attr;
+  pthread_attr_t send_attr;
+
+  log_info("", "Starting threads");
+
+  // Initialize pthread attributes (default values)
+  if (pthread_attr_init(&recv_attr))
+    log_error("TRX_ECPRI", "init pthread attributes failed\n");
+  // Set a specific stack size
+  if (pthread_attr_setstacksize(&recv_attr, PTHREAD_STACK_MIN))
+    log_error("TRX_ECPRI", "pthread setstacksize failed\n");
+  // Set scheduler policy and priority of pthread
+  if (pthread_attr_setschedpolicy(&recv_attr, SCHED_FIFO))
+    log_error("TRX_ECPRI", "pthread setschedpolicy failed\n");
+  recv_param.sched_priority = 97;
+  if (pthread_attr_setschedparam(&recv_attr, &recv_param))
+    log_error("TRX_ECPRI", "pthread setschedparam failed\n");
+  /* Use scheduling parameters of attr */
+  if (pthread_attr_setinheritsched(&recv_attr, PTHREAD_EXPLICIT_SCHED))
+    log_error("TRX_ECPRI", "pthread setinheritsched failed\n");
+
+  if (pthread_attr_init(&send_attr))
+    log_error("TRX_ECPRI", "init pthread attributes failed\n");
+  if (pthread_attr_setstacksize(&send_attr, PTHREAD_STACK_MIN))
+    log_error("TRX_ECPRI", "pthread setstacksize failed\n");
+  if (pthread_attr_setschedpolicy(&send_attr, SCHED_FIFO))
+    log_error("TRX_ECPRI", "pthread setschedpolicy failed\n");
+  send_param.sched_priority = 97;
+  if (pthread_attr_setschedparam(&send_attr, &send_param))
+    log_error("TRX_ECPRI", "pthread setschedparam failed\n");
+  if (pthread_attr_setinheritsched(&send_attr, PTHREAD_EXPLICIT_SCHED))
+    log_error("TRX_ECPRI", "pthread setinheritsched failed\n");
+
+  init_dpdk(argc, argv);
+
+  if (pthread_create(&send_pthread, NULL, send_thread, &port0))
+    error(EXIT_FAILURE, errno, "Couldn't create send thread");
+  //if (pthread_create(&send_pthread, NULL, send_thread, &port1))
+    //error(EXIT_FAILURE, errno, "Couldn't create send thread");
+  if (pthread_create(&recv_pthread, NULL, recv_thread, &port0))
+    error(EXIT_FAILURE, errno, "Couldn't create recv thread");
+  if (pthread_create(&recv_pthread, NULL, recv_thread, &port1))
+    error(EXIT_FAILURE, errno, "Couldn't create recv thread");
+
+  for(;;) {
+    sleep(1);
+  }
+
+	return 0;
+}

dpdk-test/test.sh

+#!/bin/bash
+set -e
+
+tx_packets() {
+  ethtool -S ens9f1np1|grep -v ": 0"|grep "tx_packets:"|awk '{print $2;}';
+}
+
+make;
+#A=$(tx_packets)
+nice -n -20 chrt 99 build/dpdk-recv-send -l 28 -b 0000:04:00.0 -b 0000:5e:00.0 -b 0000:5e:00.1;
+#B=$(tx_packets)
+#echo "Ethtool tx packets sent: $((B -A))";