packet: packet fanout rollover during socket overload

Changes:
  v3->v2: rebase (no other changes)
          passes selftest
  v2->v1: read f->num_members only once
          fix bug: test rollover mode + flag

Minimize packet drop in a fanout group. If one socket is full,
roll over packets to another from the group. Maintain flow
affinity during normal load using an rxhash fanout policy, while
dispersing unexpected traffic storms that hit a single cpu, such
as spoofed-source DoS flows. Rollover breaks affinity for flows
arriving at saturated sockets during those conditions.

The patch adds a fanout policy ROLLOVER that rotates between sockets,
filling each socket before moving to the next. It also adds a fanout
flag ROLLOVER. If passed along with any other fanout policy, the
primary policy is applied until the chosen socket is full. Then,
rollover selects another socket, to delay packet drop until the
entire system is saturated.

Probing sockets is not free. Selecting the last used socket, as
rollover does, is a greedy approach that maximizes chance of
success, at the cost of extreme load imbalance. In practice, with
sufficiently long queues to absorb bursts, sockets are drained in
parallel and load balance looks uniform in `top`.

To avoid contention, scales counters with number of sockets and
accesses them lockfree. Values are bounds checked to ensure
correctness.

Tested using an application with 9 threads pinned to CPUs, one socket
per thread and sufficient busywork per packet operation to limits each
thread to handling 32 Kpps. When sent 500 Kpps single UDP stream
packets, a FANOUT_CPU setup processes 32 Kpps in total without this
patch, 270 Kpps with the patch. Tested with read() and with a packet
ring (V1).

Also, passes psock_fanout.c unit test added to selftests.

Signed-off-by: Willem de Bruijn <willemb@google.com>
Reviewed-by: Eric Dumazet <edumazet@google.com>
Signed-off-by: David S. Miller <davem@davemloft.net>

7 files changed, 451 insertions, 24 deletions

diff --git a/include/uapi/linux/if_packet.h b/include/uapi/linux/if_packet.h
index f9a6037..8136658 100644
--- a/include/uapi/linux/if_packet.h
+++ b/include/uapi/linux/if_packet.h
@@ -55,6 +55,8 @@ struct sockaddr_ll {
 #define PACKET_FANOUT_HASH		0
 #define PACKET_FANOUT_LB		1
 #define PACKET_FANOUT_CPU		2
+#define PACKET_FANOUT_ROLLOVER		3
+#define PACKET_FANOUT_FLAG_ROLLOVER	0x1000
 #define PACKET_FANOUT_FLAG_DEFRAG	0x8000
 
 struct tpacket_stats {
diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
index 1d6793d..bd0d14c 100644
--- a/net/packet/af_packet.c
+++ b/net/packet/af_packet.c
@@ -181,6 +181,8 @@ static int packet_set_ring(struct sock *sk, union tpacket_req_u *req_u,
 
 struct packet_sock;
 static int tpacket_snd(struct packet_sock *po, struct msghdr *msg);
+static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
+		       struct packet_type *pt, struct net_device *orig_dev);
 
 static void *packet_previous_frame(struct packet_sock *po,
 		struct packet_ring_buffer *rb,
@@ -973,11 +975,11 @@ static void *packet_current_rx_frame(struct packet_sock *po,
 
 static void *prb_lookup_block(struct packet_sock *po,
 				     struct packet_ring_buffer *rb,
-				     unsigned int previous,
+				     unsigned int idx,
 				     int status)
 {
 	struct tpacket_kbdq_core *pkc  = GET_PBDQC_FROM_RB(rb);
-	struct tpacket_block_desc *pbd = GET_PBLOCK_DESC(pkc, previous);
+	struct tpacket_block_desc *pbd = GET_PBLOCK_DESC(pkc, idx);
 
 	if (status != BLOCK_STATUS(pbd))
 		return NULL;
@@ -1041,6 +1043,29 @@ static void packet_increment_head(struct packet_ring_buffer *buff)
 	buff->head = buff->head != buff->frame_max ? buff->head+1 : 0;
 }
 
+static bool packet_rcv_has_room(struct packet_sock *po, struct sk_buff *skb)
+{
+	struct sock *sk = &po->sk;
+	bool has_room;
+
+	if (po->prot_hook.func != tpacket_rcv)
+		return (atomic_read(&sk->sk_rmem_alloc) + skb->truesize)
+			<= sk->sk_rcvbuf;
+
+	spin_lock(&sk->sk_receive_queue.lock);
+	if (po->tp_version == TPACKET_V3)
+		has_room = prb_lookup_block(po, &po->rx_ring,
+					    po->rx_ring.prb_bdqc.kactive_blk_num,
+					    TP_STATUS_KERNEL);
+	else
+		has_room = packet_lookup_frame(po, &po->rx_ring,
+					       po->rx_ring.head,
+					       TP_STATUS_KERNEL);
+	spin_unlock(&sk->sk_receive_queue.lock);
+
+	return has_room;
+}
+
 static void packet_sock_destruct(struct sock *sk)
 {
 	skb_queue_purge(&sk->sk_error_queue);
@@ -1066,16 +1091,16 @@ static int fanout_rr_next(struct packet_fanout *f, unsigned int num)
 	return x;
 }
 
-static struct sock *fanout_demux_hash(struct packet_fanout *f, struct sk_buff *skb, unsigned int num)
+static unsigned int fanout_demux_hash(struct packet_fanout *f,
+				      struct sk_buff *skb,
+				      unsigned int num)
 {
-	u32 idx, hash = skb->rxhash;
-
-	idx = ((u64)hash * num) >> 32;
-
-	return f->arr[idx];
+	return (((u64)skb->rxhash) * num) >> 32;
 }
 
-static struct sock *fanout_demux_lb(struct packet_fanout *f, struct sk_buff *skb, unsigned int num)
+static unsigned int fanout_demux_lb(struct packet_fanout *f,
+				    struct sk_buff *skb,
+				    unsigned int num)
 {
 	int cur, old;
 
@@ -1083,14 +1108,40 @@ static struct sock *fanout_demux_lb(struct packet_fanout *f, struct sk_buff *skb
 	while ((old = atomic_cmpxchg(&f->rr_cur, cur,
 				     fanout_rr_next(f, num))) != cur)
 		cur = old;
-	return f->arr[cur];
+	return cur;
+}
+
+static unsigned int fanout_demux_cpu(struct packet_fanout *f,
+				     struct sk_buff *skb,
+				     unsigned int num)
+{
+	return smp_processor_id() % num;
 }
 
-static struct sock *fanout_demux_cpu(struct packet_fanout *f, struct sk_buff *skb, unsigned int num)
+static unsigned int fanout_demux_rollover(struct packet_fanout *f,
+					  struct sk_buff *skb,
+					  unsigned int idx, unsigned int skip,
+					  unsigned int num)
 {
-	unsigned int cpu = smp_processor_id();
+	unsigned int i, j;
 
-	return f->arr[cpu % num];
+	i = j = min_t(int, f->next[idx], num - 1);
+	do {
+		if (i != skip && packet_rcv_has_room(pkt_sk(f->arr[i]), skb)) {
+			if (i != j)
+				f->next[idx] = i;
+			return i;
+		}
+		if (++i == num)
+			i = 0;
+	} while (i != j);
+
+	return idx;
+}
+
+static bool fanout_has_flag(struct packet_fanout *f, u16 flag)
+{
+	return f->flags & (flag >> 8);
 }
 
 static int packet_rcv_fanout(struct sk_buff *skb, struct net_device *dev,
@@ -1099,7 +1150,7 @@ static int packet_rcv_fanout(struct sk_buff *skb, struct net_device *dev,
 	struct packet_fanout *f = pt->af_packet_priv;
 	unsigned int num = f->num_members;
 	struct packet_sock *po;
-	struct sock *sk;
+	unsigned int idx;
 
 	if (!net_eq(dev_net(dev), read_pnet(&f->net)) ||
 	    !num) {
@@ -1110,23 +1161,31 @@ static int packet_rcv_fanout(struct sk_buff *skb, struct net_device *dev,
 	switch (f->type) {
 	case PACKET_FANOUT_HASH:
 	default:
-		if (f->defrag) {
+		if (fanout_has_flag(f, PACKET_FANOUT_FLAG_DEFRAG)) {
 			skb = ip_check_defrag(skb, IP_DEFRAG_AF_PACKET);
 			if (!skb)
 				return 0;
 		}
 		skb_get_rxhash(skb);
-		sk = fanout_demux_hash(f, skb, num);
+		idx = fanout_demux_hash(f, skb, num);
 		break;
 	case PACKET_FANOUT_LB:
-		sk = fanout_demux_lb(f, skb, num);
+		idx = fanout_demux_lb(f, skb, num);
 		break;
 	case PACKET_FANOUT_CPU:
-		sk = fanout_demux_cpu(f, skb, num);
+		idx = fanout_demux_cpu(f, skb, num);
+		break;
+	case PACKET_FANOUT_ROLLOVER:
+		idx = fanout_demux_rollover(f, skb, 0, (unsigned int) -1, num);
 		break;
 	}
 
-	po = pkt_sk(sk);
+	po = pkt_sk(f->arr[idx]);
+	if (fanout_has_flag(f, PACKET_FANOUT_FLAG_ROLLOVER) &&
+	    unlikely(!packet_rcv_has_room(po, skb))) {
+		idx = fanout_demux_rollover(f, skb, idx, idx, num);
+		po = pkt_sk(f->arr[idx]);
+	}
 
 	return po->prot_hook.func(skb, dev, &po->prot_hook, orig_dev);
 }
@@ -1175,10 +1234,13 @@ static int fanout_add(struct sock *sk, u16 id, u16 type_flags)
 	struct packet_sock *po = pkt_sk(sk);
 	struct packet_fanout *f, *match;
 	u8 type = type_flags & 0xff;
-	u8 defrag = (type_flags & PACKET_FANOUT_FLAG_DEFRAG) ? 1 : 0;
+	u8 flags = type_flags >> 8;
 	int err;
 
 	switch (type) {
+	case PACKET_FANOUT_ROLLOVER:
+		if (type_flags & PACKET_FANOUT_FLAG_ROLLOVER)
+			return -EINVAL;
 	case PACKET_FANOUT_HASH:
 	case PACKET_FANOUT_LB:
 	case PACKET_FANOUT_CPU:
@@ -1203,7 +1265,7 @@ static int fanout_add(struct sock *sk, u16 id, u16 type_flags)
 		}
 	}
 	err = -EINVAL;
-	if (match && match->defrag != defrag)
+	if (match && match->flags != flags)
 		goto out;
 	if (!match) {
 		err = -ENOMEM;
@@ -1213,7 +1275,7 @@ static int fanout_add(struct sock *sk, u16 id, u16 type_flags)
 		write_pnet(&match->net, sock_net(sk));
 		match->id = id;
 		match->type = type;
-		match->defrag = defrag;
+		match->flags = flags;
 		atomic_set(&match->rr_cur, 0);
 		INIT_LIST_HEAD(&match->list);
 		spin_lock_init(&match->lock);
@@ -3240,7 +3302,8 @@ static int packet_getsockopt(struct socket *sock, int level, int optname,
 	case PACKET_FANOUT:
 		val = (po->fanout ?
 		       ((u32)po->fanout->id |
-			((u32)po->fanout->type << 16)) :
+			((u32)po->fanout->type << 16) |
+			((u32)po->fanout->flags << 24)) :
 		       0);
 		break;
 	case PACKET_TX_HAS_OFF:
diff --git a/net/packet/internal.h b/net/packet/internal.h
index e84cab8..e891f02 100644
--- a/net/packet/internal.h
+++ b/net/packet/internal.h
@@ -77,10 +77,11 @@ struct packet_fanout {
 	unsigned int		num_members;
 	u16			id;
 	u8			type;
-	u8			defrag;
+	u8			flags;
 	atomic_t		rr_cur;
 	struct list_head	list;
 	struct sock		*arr[PACKET_FANOUT_MAX];
+	int			next[PACKET_FANOUT_MAX];
 	spinlock_t		lock;
 	atomic_t		sk_ref;
 	struct packet_type	prot_hook ____cacheline_aligned_in_smp;
diff --git a/tools/testing/selftests/Makefile b/tools/testing/selftests/Makefile
index 7c6280f..7f50078 100644
--- a/tools/testing/selftests/Makefile
+++ b/tools/testing/selftests/Makefile
@@ -6,6 +6,7 @@ TARGETS += cpu-hotplug
 TARGETS += memory-hotplug
 TARGETS += efivarfs
 TARGETS += net-socket
+TARGETS += net-afpacket
 
 all:
 	for TARGET in $(TARGETS); do \
diff --git a/tools/testing/selftests/net-afpacket/Makefile b/tools/testing/selftests/net-afpacket/Makefile
new file mode 100644
index 0000000..45f2ffb7
--- /dev/null
+++ b/tools/testing/selftests/net-afpacket/Makefile
@@ -0,0 +1,18 @@
+# Makefile for net-socket selftests
+
+CC = $(CROSS_COMPILE)gcc
+CFLAGS = -Wall
+
+CFLAGS += -I../../../../usr/include/
+
+AF_PACKET_PROGS = psock_fanout
+
+all: $(AF_PACKET_PROGS)
+%: %.c
+	$(CC) $(CFLAGS) -o $@ $^
+
+run_tests: all
+	@/bin/sh ./run_afpackettests || echo "afpackettests: [FAIL]"
+
+clean:
+	$(RM) $(AF_PACKET_PROGS)
diff --git a/tools/testing/selftests/net-afpacket/psock_fanout.c b/tools/testing/selftests/net-afpacket/psock_fanout.c
new file mode 100644
index 0000000..09dbf93
--- /dev/null
+++ b/tools/testing/selftests/net-afpacket/psock_fanout.c
@@ -0,0 +1,326 @@
+/*
+ * Copyright 2013 Google Inc.
+ * Author: Willem de Bruijn (willemb@google.com)
+ *
+ * A basic test of packet socket fanout behavior.
+ *
+ * Control:
+ * - create fanout fails as expected with illegal flag combinations
+ * - join   fanout fails as expected with diverging types or flags
+ *
+ * Datapath:
+ *   Open a pair of packet sockets and a pair of INET sockets, send a known
+ *   number of packets across the two INET sockets and count the number of
+ *   packets enqueued onto the two packet sockets.
+ *
+ *   The test currently runs for
+ *   - PACKET_FANOUT_HASH
+ *   - PACKET_FANOUT_HASH with PACKET_FANOUT_FLAG_ROLLOVER
+ *   - PACKET_FANOUT_ROLLOVER
+ *
+ * Todo:
+ * - datapath: PACKET_FANOUT_LB
+ * - datapath: PACKET_FANOUT_CPU
+ * - functionality: PACKET_FANOUT_FLAG_DEFRAG
+ *
+ * License (GPLv2):
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms and conditions of the GNU General Public License,
+ * version 2, as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. * See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#include <arpa/inet.h>
+#include <errno.h>
+#include <linux/filter.h>
+#include <linux/if_packet.h>
+#include <net/ethernet.h>
+#include <netinet/ip.h>
+#include <netinet/udp.h>
+#include <fcntl.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+/* Hack: build even if local includes are old */
+#ifndef PACKET_FANOUT
+#define PACKET_FANOUT			18
+#define PACKET_FANOUT_HASH		0
+#define PACKET_FANOUT_LB		1
+#define PACKET_FANOUT_CPU		2
+#define PACKET_FANOUT_FLAG_DEFRAG	0x8000
+
+#ifndef PACKET_FANOUT_ROLLOVER
+#define PACKET_FANOUT_ROLLOVER		3
+#endif
+
+#ifndef PACKET_FANOUT_FLAG_ROLLOVER
+#define PACKET_FANOUT_FLAG_ROLLOVER	0x1000
+#endif
+
+#endif
+
+#define DATA_LEN			100
+#define DATA_CHAR			'a'
+
+static void pair_udp_open(int fds[], uint16_t port)
+{
+	struct sockaddr_in saddr, daddr;
+
+	fds[0] = socket(PF_INET, SOCK_DGRAM, 0);
+	fds[1] = socket(PF_INET, SOCK_DGRAM, 0);
+	if (fds[0] == -1 || fds[1] == -1) {
+		fprintf(stderr, "ERROR: socket dgram\n");
+		exit(1);
+	}
+
+	memset(&saddr, 0, sizeof(saddr));
+	saddr.sin_family = AF_INET;
+	saddr.sin_port = htons(port);
+	saddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+
+	memset(&daddr, 0, sizeof(daddr));
+	daddr.sin_family = AF_INET;
+	daddr.sin_port = htons(port + 1);
+	daddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+
+	/* must bind both to get consistent hash result */
+	if (bind(fds[1], (void *) &daddr, sizeof(daddr))) {
+		perror("bind");
+		exit(1);
+	}
+	if (bind(fds[0], (void *) &saddr, sizeof(saddr))) {
+		perror("bind");
+		exit(1);
+	}
+	if (connect(fds[0], (void *) &daddr, sizeof(daddr))) {
+		perror("bind");
+		exit(1);
+	}
+}
+
+static void pair_udp_send(int fds[], int num)
+{
+	char buf[DATA_LEN], rbuf[DATA_LEN];
+
+	memset(buf, DATA_CHAR, sizeof(buf));
+	while (num--) {
+		/* Should really handle EINTR and EAGAIN */
+		if (write(fds[0], buf, sizeof(buf)) != sizeof(buf)) {
+			fprintf(stderr, "ERROR: send failed left=%d\n", num);
+			exit(1);
+		}
+		if (read(fds[1], rbuf, sizeof(rbuf)) != sizeof(rbuf)) {
+			fprintf(stderr, "ERROR: recv failed left=%d\n", num);
+			exit(1);
+		}
+		if (memcmp(buf, rbuf, sizeof(buf))) {
+			fprintf(stderr, "ERROR: data failed left=%d\n", num);
+			exit(1);
+		}
+	}
+}
+
+static void sock_fanout_setfilter(int fd)
+{
+	struct sock_filter bpf_filter[] = {
+		{ 0x80, 0, 0, 0x00000000 },  /* LD  pktlen		      */
+		{ 0x35, 0, 5, DATA_LEN   },  /* JGE DATA_LEN  [f goto nomatch]*/
+		{ 0x30, 0, 0, 0x00000050 },  /* LD  ip[80]		      */
+		{ 0x15, 0, 3, DATA_CHAR  },  /* JEQ DATA_CHAR [f goto nomatch]*/
+		{ 0x30, 0, 0, 0x00000051 },  /* LD  ip[81]		      */
+		{ 0x15, 0, 1, DATA_CHAR  },  /* JEQ DATA_CHAR [f goto nomatch]*/
+		{ 0x6, 0, 0, 0x00000060  },  /* RET match	              */
+/* nomatch */	{ 0x6, 0, 0, 0x00000000  },  /* RET no match		      */
+	};
+	struct sock_fprog bpf_prog;
+
+	bpf_prog.filter = bpf_filter;
+	bpf_prog.len = sizeof(bpf_filter) / sizeof(struct sock_filter);
+	if (setsockopt(fd, SOL_SOCKET, SO_ATTACH_FILTER, &bpf_prog,
+		       sizeof(bpf_prog))) {
+		perror("setsockopt SO_ATTACH_FILTER");
+		exit(1);
+	}
+}
+
+/* Open a socket in a given fanout mode.
+ * @return -1 if mode is bad, a valid socket otherwise */
+static int sock_fanout_open(uint16_t typeflags, int num_packets)
+{
+	int fd, val;
+
+	fd = socket(PF_PACKET, SOCK_DGRAM, htons(ETH_P_IP));
+	if (fd < 0) {
+		perror("socket packet");
+		exit(1);
+	}
+
+	/* fanout group ID is always 0: tests whether old groups are deleted */
+	val = ((int) typeflags) << 16;
+	if (setsockopt(fd, SOL_PACKET, PACKET_FANOUT, &val, sizeof(val))) {
+		if (close(fd)) {
+			perror("close packet");
+			exit(1);
+		}
+		return -1;
+	}
+
+	val = sizeof(struct iphdr) + sizeof(struct udphdr) + DATA_LEN;
+	val *= num_packets;
+	/* hack: apparently, the above calculation is too small (TODO: fix) */
+	val *= 3;
+	if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val))) {
+		perror("setsockopt SO_RCVBUF");
+		exit(1);
+	}
+
+	sock_fanout_setfilter(fd);
+	return fd;
+}
+
+static void sock_fanout_read(int fds[], const int expect[])
+{
+	struct tpacket_stats stats;
+	socklen_t ssize;
+	int ret[2];
+
+	ssize = sizeof(stats);
+	if (getsockopt(fds[0], SOL_PACKET, PACKET_STATISTICS, &stats, &ssize)) {
+		perror("getsockopt statistics 0");
+		exit(1);
+	}
+	ret[0] = stats.tp_packets - stats.tp_drops;
+	ssize = sizeof(stats);
+	if (getsockopt(fds[1], SOL_PACKET, PACKET_STATISTICS, &stats, &ssize)) {
+		perror("getsockopt statistics 1");
+		exit(1);
+	}
+	ret[1] = stats.tp_packets - stats.tp_drops;
+
+	fprintf(stderr, "info: count=%d,%d, expect=%d,%d\n",
+			ret[0], ret[1], expect[0], expect[1]);
+
+	if ((!(ret[0] == expect[0] && ret[1] == expect[1])) &&
+	    (!(ret[0] == expect[1] && ret[1] == expect[0]))) {
+		fprintf(stderr, "ERROR: incorrect queue lengths\n");
+		exit(1);
+	}
+}
+
+/* Test illegal mode + flag combination */
+static void test_control_single(void)
+{
+	fprintf(stderr, "test: control single socket\n");
+
+	if (sock_fanout_open(PACKET_FANOUT_ROLLOVER |
+			       PACKET_FANOUT_FLAG_ROLLOVER, 0) != -1) {
+		fprintf(stderr, "ERROR: opened socket with dual rollover\n");
+		exit(1);
+	}
+}
+
+/* Test illegal group with different modes or flags */
+static void test_control_group(void)
+{
+	int fds[2];
+
+	fprintf(stderr, "test: control multiple sockets\n");
+
+	fds[0] = sock_fanout_open(PACKET_FANOUT_HASH, 20);
+	if (fds[0] == -1) {
+		fprintf(stderr, "ERROR: failed to open HASH socket\n");
+		exit(1);
+	}
+	if (sock_fanout_open(PACKET_FANOUT_HASH |
+			       PACKET_FANOUT_FLAG_DEFRAG, 10) != -1) {
+		fprintf(stderr, "ERROR: joined group with wrong flag defrag\n");
+		exit(1);
+	}
+	if (sock_fanout_open(PACKET_FANOUT_HASH |
+			       PACKET_FANOUT_FLAG_ROLLOVER, 10) != -1) {
+		fprintf(stderr, "ERROR: joined group with wrong flag ro\n");
+		exit(1);
+	}
+	if (sock_fanout_open(PACKET_FANOUT_CPU, 10) != -1) {
+		fprintf(stderr, "ERROR: joined group with wrong mode\n");
+		exit(1);
+	}
+	fds[1] = sock_fanout_open(PACKET_FANOUT_HASH, 20);
+	if (fds[1] == -1) {
+		fprintf(stderr, "ERROR: failed to join group\n");
+		exit(1);
+	}
+	if (close(fds[1]) || close(fds[0])) {
+		fprintf(stderr, "ERROR: closing sockets\n");
+		exit(1);
+	}
+}
+
+static void test_datapath(uint16_t typeflags,
+			  const int expect1[], const int expect2[])
+{
+	const int expect0[] = { 0, 0 };
+	int fds[2], fds_udp[2][2];
+
+	fprintf(stderr, "test: datapath 0x%hx\n", typeflags);
+
+	fds[0] = sock_fanout_open(typeflags, 20);
+	fds[1] = sock_fanout_open(typeflags, 20);
+	if (fds[0] == -1 || fds[1] == -1) {
+		fprintf(stderr, "ERROR: failed open\n");
+		exit(1);
+	}
+	pair_udp_open(fds_udp[0], 8000);
+	pair_udp_open(fds_udp[1], 8002);
+	sock_fanout_read(fds, expect0);
+
+	/* Send data, but not enough to overflow a queue */
+	pair_udp_send(fds_udp[0], 15);
+	pair_udp_send(fds_udp[1], 5);
+	sock_fanout_read(fds, expect1);
+
+	/* Send more data, overflow the queue */
+	pair_udp_send(fds_udp[0], 15);
+	/* TODO: ensure consistent order between expect1 and expect2 */
+	sock_fanout_read(fds, expect2);
+
+	if (close(fds_udp[1][1]) || close(fds_udp[1][0]) ||
+	    close(fds_udp[0][1]) || close(fds_udp[0][0]) ||
+	    close(fds[1]) || close(fds[0])) {
+		fprintf(stderr, "close datapath\n");
+		exit(1);
+	}
+}
+
+int main(int argc, char **argv)
+{
+	const int expect_hash[2][2]	= { { 15, 5 }, { 5, 0 } };
+	const int expect_hash_rb[2][2]	= { { 15, 5 }, { 5, 10 } };
+	const int expect_rb[2][2]	= { { 20, 0 }, { 0, 15 } };
+
+	test_control_single();
+	test_control_group();
+
+	test_datapath(PACKET_FANOUT_HASH, expect_hash[0], expect_hash[1]);
+	test_datapath(PACKET_FANOUT_HASH | PACKET_FANOUT_FLAG_ROLLOVER,
+		      expect_hash_rb[0], expect_hash_rb[1]);
+	test_datapath(PACKET_FANOUT_ROLLOVER, expect_rb[0], expect_rb[1]);
+
+	printf("OK. All tests passed\n");
+	return 0;
+}
diff --git a/tools/testing/selftests/net-afpacket/run_afpackettests b/tools/testing/selftests/net-afpacket/run_afpackettests
new file mode 100644
index 0000000..7907824
--- /dev/null
+++ b/tools/testing/selftests/net-afpacket/run_afpackettests
@@ -0,0 +1,16 @@
+#!/bin/sh
+
+if [ $(id -u) != 0 ]; then
+	echo $msg must be run as root >&2
+	exit 0
+fi
+
+echo "--------------------"
+echo "running psock_fanout test"
+echo "--------------------"
+./psock_fanout
+if [ $? -ne 0 ]; then
+	echo "[FAIL]"
+else
+	echo "[PASS]"
+fi