Merge branch 'af_xdp-stats'

Ciara Loftus says:

====================
This series introduces new statistics for af_xdp:
1. drops due to rx ring being full
2. drops due to fill ring being empty
3. failures pulling an item from the tx ring

These statistics should assist users debugging and troubleshooting
peformance issues and packet drops.

The statistics are made available though the getsockopt and xsk_diag
interfaces, and the ability to dump these extended statistics is made
available in the xdpsock application via the --extra-stats or -x flag.

A separate patch which will add ss/iproute2 support will follow.
====================

Acked-by: Björn Töpel <bjorn.topel@intel.com>
Signed-off-by: Alexei Starovoitov <ast@kernel.org>

include/net/xdp_sock.h

@@ -69,7 +69,11 @@ struct xdp_sock {
 	spinlock_t tx_completion_lock;
 	/* Protects generic receive. */
 	spinlock_t rx_lock;
+
+	/* Statistics */
 	u64 rx_dropped;
+	u64 rx_queue_full;
+
 	struct list_head map_list;
 	/* Protects map_list */
 	spinlock_t map_list_lock;

include/uapi/linux/if_xdp.h

@@ -73,9 +73,12 @@ struct xdp_umem_reg {
 };
 
 struct xdp_statistics {
-	__u64 rx_dropped; /* Dropped for reasons other than invalid desc */
+	__u64 rx_dropped; /* Dropped for other reasons */
 	__u64 rx_invalid_descs; /* Dropped due to invalid descriptor */
 	__u64 tx_invalid_descs; /* Dropped due to invalid descriptor */
+	__u64 rx_ring_full; /* Dropped due to rx ring being full */
+	__u64 rx_fill_ring_empty_descs; /* Failed to retrieve item from fill ring */
+	__u64 tx_ring_empty_descs; /* Failed to retrieve item from tx ring */
 };
 
 struct xdp_options {

include/uapi/linux/xdp_diag.h

@@ -30,6 +30,7 @@ struct xdp_diag_msg {
 #define XDP_SHOW_RING_CFG	(1 << 1)
 #define XDP_SHOW_UMEM		(1 << 2)
 #define XDP_SHOW_MEMINFO	(1 << 3)
+#define XDP_SHOW_STATS		(1 << 4)
 
 enum {
 	XDP_DIAG_NONE,
@@ -41,6 +42,7 @@ enum {
 	XDP_DIAG_UMEM_FILL_RING,
 	XDP_DIAG_UMEM_COMPLETION_RING,
 	XDP_DIAG_MEMINFO,
+	XDP_DIAG_STATS,
 	__XDP_DIAG_MAX,
 };
 
@@ -69,4 +71,13 @@ struct xdp_diag_umem {
 	__u32	refs;
 };
 
+struct xdp_diag_stats {
+	__u64	n_rx_dropped;
+	__u64	n_rx_invalid;
+	__u64	n_rx_full;
+	__u64	n_fill_ring_empty;
+	__u64	n_tx_invalid;
+	__u64	n_tx_ring_empty;
+};
+
 #endif /* _LINUX_XDP_DIAG_H */

net/xdp/xsk.c

@@ -123,7 +123,7 @@ static int __xsk_rcv_zc(struct xdp_sock *xs, struct xdp_buff *xdp, u32 len)
 	addr = xp_get_handle(xskb);
 	err = xskq_prod_reserve_desc(xs->rx, addr, len);
 	if (err) {
-		xs->rx_dropped++;
+		xs->rx_queue_full++;
 		return err;
 	}
 
@@ -274,8 +274,10 @@ bool xsk_umem_consume_tx(struct xdp_umem *umem, struct xdp_desc *desc)
 
 	rcu_read_lock();
 	list_for_each_entry_rcu(xs, &umem->xsk_tx_list, list) {
-		if (!xskq_cons_peek_desc(xs->tx, desc, umem))
+		if (!xskq_cons_peek_desc(xs->tx, desc, umem)) {
+			xs->tx->queue_empty_descs++;
 			continue;
+		}
 
 		/* This is the backpressure mechanism for the Tx path.
 		 * Reserve space in the completion queue and only proceed
@@ -387,6 +389,8 @@ static int xsk_generic_xmit(struct sock *sk)
 		sent_frame = true;
 	}
 
+	xs->tx->queue_empty_descs++;
+
 out:
 	if (sent_frame)
 		sk->sk_write_space(sk);
@@ -812,6 +816,12 @@ static void xsk_enter_umem_offsets(struct xdp_ring_offset_v1 *ring)
 	ring->desc = offsetof(struct xdp_umem_ring, desc);
 }
 
+struct xdp_statistics_v1 {
+	__u64 rx_dropped;
+	__u64 rx_invalid_descs;
+	__u64 tx_invalid_descs;
+};
+
 static int xsk_getsockopt(struct socket *sock, int level, int optname,
 			  char __user *optval, int __user *optlen)
 {
@@ -831,19 +841,35 @@ static int xsk_getsockopt(struct socket *sock, int level, int optname,
 	case XDP_STATISTICS:
 	{
 		struct xdp_statistics stats;
+		bool extra_stats = true;
+		size_t stats_size;
 
-		if (len < sizeof(stats))
+		if (len < sizeof(struct xdp_statistics_v1)) {
 			return -EINVAL;
+		} else if (len < sizeof(stats)) {
+			extra_stats = false;
+			stats_size = sizeof(struct xdp_statistics_v1);
+		} else {
+			stats_size = sizeof(stats);
+		}
 
 		mutex_lock(&xs->mutex);
 		stats.rx_dropped = xs->rx_dropped;
+		if (extra_stats) {
+			stats.rx_ring_full = xs->rx_queue_full;
+			stats.rx_fill_ring_empty_descs =
+				xs->umem ? xskq_nb_queue_empty_descs(xs->umem->fq) : 0;
+			stats.tx_ring_empty_descs = xskq_nb_queue_empty_descs(xs->tx);
+		} else {
+			stats.rx_dropped += xs->rx_queue_full;
+		}
 		stats.rx_invalid_descs = xskq_nb_invalid_descs(xs->rx);
 		stats.tx_invalid_descs = xskq_nb_invalid_descs(xs->tx);
 		mutex_unlock(&xs->mutex);
 
-		if (copy_to_user(optval, &stats, sizeof(stats)))
+		if (copy_to_user(optval, &stats, stats_size))
 			return -EFAULT;
-		if (put_user(sizeof(stats), optlen))
+		if (put_user(stats_size, optlen))
 			return -EFAULT;
 
 		return 0;

net/xdp/xsk_buff_pool.c

@@ -235,6 +235,7 @@ static struct xdp_buff_xsk *__xp_alloc(struct xsk_buff_pool *pool)
 
 	for (;;) {
 		if (!xskq_cons_peek_addr_unchecked(pool->fq, &addr)) {
+			pool->fq->queue_empty_descs++;
 			xp_release(xskb);
 			return NULL;
 		}

net/xdp/xsk_diag.c

@@ -76,6 +76,19 @@ static int xsk_diag_put_umem(const struct xdp_sock *xs, struct sk_buff *nlskb)
 	return err;
 }
 
+static int xsk_diag_put_stats(const struct xdp_sock *xs, struct sk_buff *nlskb)
+{
+	struct xdp_diag_stats du = {};
+
+	du.n_rx_dropped = xs->rx_dropped;
+	du.n_rx_invalid = xskq_nb_invalid_descs(xs->rx);
+	du.n_rx_full = xs->rx_queue_full;
+	du.n_fill_ring_empty = xs->umem ? xskq_nb_queue_empty_descs(xs->umem->fq) : 0;
+	du.n_tx_invalid = xskq_nb_invalid_descs(xs->tx);
+	du.n_tx_ring_empty = xskq_nb_queue_empty_descs(xs->tx);
+	return nla_put(nlskb, XDP_DIAG_STATS, sizeof(du), &du);
+}
+
 static int xsk_diag_fill(struct sock *sk, struct sk_buff *nlskb,
 			 struct xdp_diag_req *req,
 			 struct user_namespace *user_ns,
@@ -118,6 +131,10 @@ static int xsk_diag_fill(struct sock *sk, struct sk_buff *nlskb,
 	    sock_diag_put_meminfo(sk, nlskb, XDP_DIAG_MEMINFO))
 		goto out_nlmsg_trim;
 
+	if ((req->xdiag_show & XDP_SHOW_STATS) &&
+	    xsk_diag_put_stats(xs, nlskb))
+		goto out_nlmsg_trim;
+
 	mutex_unlock(&xs->mutex);
 	nlmsg_end(nlskb, nlh);
 	return 0;

net/xdp/xsk_queue.h

@@ -38,6 +38,7 @@ struct xsk_queue {
 	u32 cached_cons;
 	struct xdp_ring *ring;
 	u64 invalid_descs;
+	u64 queue_empty_descs;
 };
 
 /* The structure of the shared state of the rings are the same as the
@@ -354,6 +355,11 @@ static inline u64 xskq_nb_invalid_descs(struct xsk_queue *q)
 	return q ? q->invalid_descs : 0;
 }
 
+static inline u64 xskq_nb_queue_empty_descs(struct xsk_queue *q)
+{
+	return q ? q->queue_empty_descs : 0;
+}
+
 struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
 void xskq_destroy(struct xsk_queue *q_ops);
 

samples/bpf/xdpsock_user.c

@@ -77,6 +77,7 @@ static u32 opt_batch_size = 64;
 static int opt_pkt_count;
 static u16 opt_pkt_size = MIN_PKT_SIZE;
 static u32 opt_pkt_fill_pattern = 0x12345678;
+static bool opt_extra_stats;
 static int opt_poll;
 static int opt_interval = 1;
 static u32 opt_xdp_bind_flags = XDP_USE_NEED_WAKEUP;
@@ -103,8 +104,20 @@ struct xsk_socket_info {
 	struct xsk_socket *xsk;
 	unsigned long rx_npkts;
 	unsigned long tx_npkts;
+	unsigned long rx_dropped_npkts;
+	unsigned long rx_invalid_npkts;
+	unsigned long tx_invalid_npkts;
+	unsigned long rx_full_npkts;
+	unsigned long rx_fill_empty_npkts;
+	unsigned long tx_empty_npkts;
 	unsigned long prev_rx_npkts;
 	unsigned long prev_tx_npkts;
+	unsigned long prev_rx_dropped_npkts;
+	unsigned long prev_rx_invalid_npkts;
+	unsigned long prev_tx_invalid_npkts;
+	unsigned long prev_rx_full_npkts;
+	unsigned long prev_rx_fill_empty_npkts;
+	unsigned long prev_tx_empty_npkts;
 	u32 outstanding_tx;
 };
 
@@ -147,6 +160,30 @@ static void print_benchmark(bool running)
 	}
 }
 
+static int xsk_get_xdp_stats(int fd, struct xsk_socket_info *xsk)
+{
+	struct xdp_statistics stats;
+	socklen_t optlen;
+	int err;
+
+	optlen = sizeof(stats);
+	err = getsockopt(fd, SOL_XDP, XDP_STATISTICS, &stats, &optlen);
+	if (err)
+		return err;
+
+	if (optlen == sizeof(struct xdp_statistics)) {
+		xsk->rx_dropped_npkts = stats.rx_dropped;
+		xsk->rx_invalid_npkts = stats.rx_invalid_descs;
+		xsk->tx_invalid_npkts = stats.tx_invalid_descs;
+		xsk->rx_full_npkts = stats.rx_ring_full;
+		xsk->rx_fill_empty_npkts = stats.rx_fill_ring_empty_descs;
+		xsk->tx_empty_npkts = stats.tx_ring_empty_descs;
+		return 0;
+	}
+
+	return -EINVAL;
+}
+
 static void dump_stats(void)
 {
 	unsigned long now = get_nsecs();
@@ -157,7 +194,8 @@ static void dump_stats(void)
 
 	for (i = 0; i < num_socks && xsks[i]; i++) {
 		char *fmt = "%-15s %'-11.0f %'-11lu\n";
-		double rx_pps, tx_pps;
+		double rx_pps, tx_pps, dropped_pps, rx_invalid_pps, full_pps, fill_empty_pps,
+			tx_invalid_pps, tx_empty_pps;
 
 		rx_pps = (xsks[i]->rx_npkts - xsks[i]->prev_rx_npkts) *
 			 1000000000. / dt;
@@ -175,6 +213,46 @@ static void dump_stats(void)
 
 		xsks[i]->prev_rx_npkts = xsks[i]->rx_npkts;
 		xsks[i]->prev_tx_npkts = xsks[i]->tx_npkts;
+
+		if (opt_extra_stats) {
+			if (!xsk_get_xdp_stats(xsk_socket__fd(xsks[i]->xsk), xsks[i])) {
+				dropped_pps = (xsks[i]->rx_dropped_npkts -
+						xsks[i]->prev_rx_dropped_npkts) * 1000000000. / dt;
+				rx_invalid_pps = (xsks[i]->rx_invalid_npkts -
+						xsks[i]->prev_rx_invalid_npkts) * 1000000000. / dt;
+				tx_invalid_pps = (xsks[i]->tx_invalid_npkts -
+						xsks[i]->prev_tx_invalid_npkts) * 1000000000. / dt;
+				full_pps = (xsks[i]->rx_full_npkts -
+						xsks[i]->prev_rx_full_npkts) * 1000000000. / dt;
+				fill_empty_pps = (xsks[i]->rx_fill_empty_npkts -
+						xsks[i]->prev_rx_fill_empty_npkts)
+						* 1000000000. / dt;
+				tx_empty_pps = (xsks[i]->tx_empty_npkts -
+						xsks[i]->prev_tx_empty_npkts) * 1000000000. / dt;
+
+				printf(fmt, "rx dropped", dropped_pps,
+				       xsks[i]->rx_dropped_npkts);
+				printf(fmt, "rx invalid", rx_invalid_pps,
+				       xsks[i]->rx_invalid_npkts);
+				printf(fmt, "tx invalid", tx_invalid_pps,
+				       xsks[i]->tx_invalid_npkts);
+				printf(fmt, "rx queue full", full_pps,
+				       xsks[i]->rx_full_npkts);
+				printf(fmt, "fill ring empty", fill_empty_pps,
+				       xsks[i]->rx_fill_empty_npkts);
+				printf(fmt, "tx ring empty", tx_empty_pps,
+				       xsks[i]->tx_empty_npkts);
+
+				xsks[i]->prev_rx_dropped_npkts = xsks[i]->rx_dropped_npkts;
+				xsks[i]->prev_rx_invalid_npkts = xsks[i]->rx_invalid_npkts;
+				xsks[i]->prev_tx_invalid_npkts = xsks[i]->tx_invalid_npkts;
+				xsks[i]->prev_rx_full_npkts = xsks[i]->rx_full_npkts;
+				xsks[i]->prev_rx_fill_empty_npkts = xsks[i]->rx_fill_empty_npkts;
+				xsks[i]->prev_tx_empty_npkts = xsks[i]->tx_empty_npkts;
+			} else {
+				printf("%-15s\n", "Error retrieving extra stats");
+			}
+		}
 	}
 }
 
@@ -630,6 +708,7 @@ static struct option long_options[] = {
 	{"tx-pkt-count", required_argument, 0, 'C'},
 	{"tx-pkt-size", required_argument, 0, 's'},
 	{"tx-pkt-pattern", required_argument, 0, 'P'},
+	{"extra-stats", no_argument, 0, 'x'},
 	{0, 0, 0, 0}
 };
 
@@ -664,6 +743,7 @@ static void usage(const char *prog)
 		"			(Default: %d bytes)\n"
 		"			Min size: %d, Max size %d.\n"
 		"  -P, --tx-pkt-pattern=nPacket fill pattern. Default: 0x%x\n"
+		"  -x, --extra-stats	Display extra statistics.\n"
 		"\n";
 	fprintf(stderr, str, prog, XSK_UMEM__DEFAULT_FRAME_SIZE,
 		opt_batch_size, MIN_PKT_SIZE, MIN_PKT_SIZE,
@@ -679,7 +759,7 @@ static void parse_command_line(int argc, char **argv)
 	opterr = 0;
 
 	for (;;) {
-		c = getopt_long(argc, argv, "Frtli:q:pSNn:czf:muMd:b:C:s:P:",
+		c = getopt_long(argc, argv, "Frtli:q:pSNn:czf:muMd:b:C:s:P:x",
 				long_options, &option_index);
 		if (c == -1)
 			break;
@@ -760,6 +840,9 @@ static void parse_command_line(int argc, char **argv)
 		case 'P':
 			opt_pkt_fill_pattern = strtol(optarg, NULL, 16);
 			break;
+		case 'x':
+			opt_extra_stats = 1;
+			break;
 		default:
 			usage(basename(argv[0]));
 		}

tools/include/uapi/linux/if_xdp.h

@@ -73,9 +73,12 @@ struct xdp_umem_reg {
 };
 
 struct xdp_statistics {
-	__u64 rx_dropped; /* Dropped for reasons other than invalid desc */
+	__u64 rx_dropped; /* Dropped for other reasons */
 	__u64 rx_invalid_descs; /* Dropped due to invalid descriptor */
 	__u64 tx_invalid_descs; /* Dropped due to invalid descriptor */
+	__u64 rx_ring_full; /* Dropped due to rx ring being full */
+	__u64 rx_fill_ring_empty_descs; /* Failed to retrieve item from fill ring */
+	__u64 tx_ring_empty_descs; /* Failed to retrieve item from tx ring */
 };
 
 struct xdp_options {