1.3M TX PPS with XDP

af-xdp/test-xdp.sh

+#!/bin/bash
+set -e
+
+xdp_off() {
+  ip link set dev ens9f1np1 xdp off
+}
+tx_packets() {
+  ethtool -S ens9f1np1|grep -v ": 0"|grep "tx_packets:"|awk '{print $2;}';
+}
+
+xdp_off;
+make clean && make;
+A=$(tx_packets)
+nice -n -20 chrt 99 ./xdp_user;
+B=$(tx_packets)
+echo "Ethtool tx packets sent: $((B -A))";
+xdp_off;

af-xdp/xdp_user.c

@@ -47,9 +47,16 @@
 #define err_errno(...) error(EXIT_FAILURE, errno, __VA_ARGS__);
 #define NUM_FRAMES 4096
 #define ETH_FCS_SIZE 4
-#define FRAME_SIZE XSK_UMEM__DEFAULT_FRAME_SIZE
-#define PACKET_SIZE 64
-#define DEBUG
+//#define FRAME_SIZE XSK_UMEM__DEFAULT_FRAME_SIZE
+#define FRAME_SIZE 2048
+#define PACKET_SIZE 262
+//#define NB_PACKETS 1000000
+#define NB_PACKETS 100
+#define BATCH_SIZE 2048
+#define PRINT_PROGRESS
+#define PRINT_PROGRESS_INTERVAL 1000
+
+//#define DEBUG
 static void log_info(const char * section, const char * msg, ...) {
   time_t t;
   struct tm ts;
@@ -97,6 +104,30 @@ struct xdpsock {
 	struct xsk_socket *xsk;
 	int fd;
 };
+// 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;
+}
 
 static void init_xdp_recv(char * network_if);
 static void init_xdp_send(char * network_if);
@@ -106,11 +137,14 @@ static void send_xdp_packet(void);
 static void recv_xdp_cleanup(void);
 static void close_xdp_socket(void);
 
-static uint8_t pkt_data[XSK_UMEM__DEFAULT_FRAME_SIZE];
+static uint8_t pkt_data[FRAME_SIZE];
 static struct xdpsock send_xdp_socket;
 static struct xdpsock recv_xdp_socket;
 static int received;
+static int64_t sent;
 static uint32_t idx_rx = 0, idx_recv;
+static int batch_size = BATCH_SIZE;
+int trace_fd;
 
 int main() {
   char * network_if = "ens9f1np1";
@@ -122,10 +156,53 @@ int main() {
   return 0;
 }
 
+void trace_write(const char *fmt, ...)
+{
+    va_list ap;
+    char buf[256];
+    int n;
+
+    if (trace_fd < 0)
+        return;
+    va_start(ap, fmt);
+    n = vsnprintf(buf, 256, fmt, ap);
+    va_end(ap);
+    write(trace_fd, buf, n);
+}
+
+static int latency_target_fd = -1;
+static int32_t latency_target_value = 0;
+void set_latency_target(void) {
+  struct stat s;
+  int err;
+
+  errno = 0;
+  err = stat("/dev/cpu_dma_latency", &s);
+  if (err == -1) {
+    error(EXIT_FAILURE, errno, "WARN: stat /dev/cpu_dma_latency failed");
+    return;
+  }
+
+  errno = 0;
+  latency_target_fd = open("/dev/cpu_dma_latency", O_RDWR);
+  if (latency_target_fd == -1) {
+    error(EXIT_FAILURE, errno, "WARN: open /dev/cpu_dma_latency");
+    return;
+  }
+
+  errno = 0;
+  err = write(latency_target_fd, &latency_target_value, 4);
+  if (err < 1) {
+    error(EXIT_FAILURE, errno, "# error setting cpu_dma_latency to %d!",
+          latency_target_value);
+    close(latency_target_fd);
+    return;
+  }
+  printf("# /dev/cpu_dma_latency set to %dus\n", latency_target_value);
+}
 static void init_xdp_send(char * network_if) {
   uint32_t idx;
   int ret, i, xsks_map = 0;
-  int batch_size;
   struct xsk_socket_config xsk_cfg;
   struct xsk_umem_config cfg = {
       .fill_size = XSK_RING_PROD__DEFAULT_NUM_DESCS,
@@ -137,10 +214,24 @@ static void init_xdp_send(char * network_if) {
   void *buffer = NULL;
   struct ethhdr *eth_hdr = (struct ethhdr *)pkt_data;
 
+	//set_latency_target();
+  if (mlockall(MCL_CURRENT | MCL_FUTURE)) {
+    fprintf(stderr, "mlockall failed");
+    exit(EXIT_FAILURE);
+  }
+
+	for(int j = 0; j < PACKET_SIZE; j++)
+		pkt_data[j] = 0x17;
+
   memcpy(eth_hdr->h_dest, "\xb8\x59\x9f\x07\x7d\xdb", ETH_ALEN);
   memcpy(eth_hdr->h_source, "\x04\x09\xa5\x0f\x9f\x4c", ETH_ALEN);
   eth_hdr->h_proto = htons(ETH_P_IP);
 
+	//for(int j = 0; j < PACKET_SIZE; j++)
+	//	printf("%x", pkt_data[j]);
+	//printf("\n");
+	//exit(EXIT_SUCCESS);
+
   log_debug("", "posix_memalign");
   /* Allocate user space memory for xdp frames */
   ret = posix_memalign(&buffer, sysconf(_SC_PAGE_SIZE), NUM_FRAMES * FRAME_SIZE);
@@ -159,7 +250,7 @@ static void init_xdp_send(char * network_if) {
   xsk_cfg.tx_size = XSK_RING_PROD__DEFAULT_NUM_DESCS;
   xsk_cfg.libbpf_flags = 0;
   xsk_cfg.xdp_flags = XDP_FLAGS_DRV_MODE;
-  xsk_cfg.bind_flags = 0;
+  xsk_cfg.bind_flags = XDP_USE_NEED_WAKEUP;
 
   log_debug("", "xsk_socket__create");
   ret = xsk_socket__create(&send_xdp_socket.xsk, network_if, 0, send_xdp_socket.umem.umem,
@@ -171,7 +262,6 @@ static void init_xdp_send(char * network_if) {
 static void init_xdp_recv(char * network_if) {
   uint32_t idx;
   int ret, i, xsks_map = 0;
-  int batch_size;
   struct xsk_socket_config xsk_cfg;
   struct xsk_umem_config cfg = {
       .fill_size = XSK_RING_PROD__DEFAULT_NUM_DESCS,
@@ -181,6 +271,7 @@ static void init_xdp_recv(char * network_if) {
       .flags = 0,
   };
   void *buffer = NULL;
+  sent = 0;
 
   log_debug("", "posix_memalign");
   /* Allocate user space memory for xdp frames */
@@ -209,38 +300,64 @@ static void init_xdp_recv(char * network_if) {
   log_debug("", "recv_xdp_packet");
   recv_xdp_packet();
 }
+static void complete_tx_only(void) {
+  int rcvd;
+  uint32_t idx;
+  rcvd = xsk_ring_cons__peek(&send_xdp_socket.umem.cq, batch_size, &idx);
+  if (rcvd > 0) {
+    xsk_ring_cons__release(&send_xdp_socket.umem.cq, rcvd);
+    sent += rcvd;
+  }
+}
+
 static void send_xdp_packet(void) {
   struct pollfd fds[1] = {};
   int i, ret;
+  struct timespec start, end;
+  int64_t duration_ns;
+#ifdef PRINT_PROGRESS
+	int64_t progress = PRINT_PROGRESS_INTERVAL;
+#endif
   fds[0].fd = xsk_socket__fd(send_xdp_socket.xsk);
   fds[0].events = POLLOUT;
 
   for (int i = 0; i < NUM_FRAMES; i++)
-    memcpy(xsk_umem__get_data(send_xdp_socket.umem.buffer, i * XSK_UMEM__DEFAULT_FRAME_SIZE), pkt_data, PACKET_SIZE - ETH_FCS_SIZE);
+    memcpy(xsk_umem__get_data(send_xdp_socket.umem.buffer, i * FRAME_SIZE), pkt_data, PACKET_SIZE - ETH_FCS_SIZE);
 
-  while(1) {
-    int batch_size = 64;
+  clock_gettime(CLOCK_MONOTONIC, &start);
+  for(int l = 0; (sent < NB_PACKETS); l++) {
     uint32_t idx;
 
-    log_debug("", "polling");
-    ret = poll(fds, 1, -1);
-    if ((ret <= 0) || !(fds[0].revents & POLLOUT))
+#ifdef PRINT_PROGRESS
+		if(sent > progress) {
+			printf("%" PRIi64 "\n", sent);
+			progress += PRINT_PROGRESS_INTERVAL;
+		}
+#endif
+    ret = poll(fds, 1, 0);
+    if ((ret <= 0) || !(fds[0].revents & POLLOUT)) {
+			complete_tx_only();
       continue;
+		}
 
-    log_debug("", "reserve");
-    xsk_ring_prod__reserve(&send_xdp_socket.tx, batch_size, &idx);
+    while(xsk_ring_prod__reserve(&send_xdp_socket.tx, batch_size, &idx) < batch_size)
+      complete_tx_only();
 
     for (int k = 0; k < batch_size; k++) {
       struct xdp_desc *tx_desc = xsk_ring_prod__tx_desc(&send_xdp_socket.tx, idx + k);
-      tx_desc->addr = k * (PACKET_SIZE);
+      tx_desc->addr = k * (FRAME_SIZE);
       tx_desc->len = PACKET_SIZE - ETH_FCS_SIZE;
     }
 
-    log_debug("", "submit");
     xsk_ring_prod__submit(&send_xdp_socket.tx, batch_size);
-    log_debug("", "submit done");
-    break;
+    if (xsk_ring_prod__needs_wakeup(&send_xdp_socket.tx))
+          sendto(xsk_socket__fd(send_xdp_socket.xsk), NULL, 0, MSG_DONTWAIT, NULL, 0);
   }
+  clock_gettime(CLOCK_MONOTONIC, &end);
+  duration_ns = calcdiff_ns(end, start);
+  log_info("STATS", "Duration: %" PRIi64 " us", duration_ns / 1000);
+  log_info("STATS", "Packets sent: %" PRIi64 " / %" PRIi64, sent, NB_PACKETS);
+  log_info("STATS", "pps: %" PRIi64, (sent * NSEC_PER_SEC) / duration_ns);
 }
 
 static int recv_xdp_packet(void) {