iv/linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

packet: packet fanout rollover during socket overload

Browse Source 
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>
This commit is contained in:
Willem de Bruijn 2013-03-19 10:18:11 +00:00 committed by David S. Miller
parent b0aa73bf08
commit 77f65ebdca
7 changed files with 451 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/uapi/linux/if_packet.h
View File

@ -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 {

109
net/packet/af_packet.c
View File

@ -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 struct sock *fanout_demux_cpu(struct packet_fanout *f, struct sk_buff *skb, unsigned int num)
static unsigned int fanout_demux_cpu(struct packet_fanout *f,
struct sk_buff *skb,
unsigned int num)
{
unsigned int cpu = smp_processor_id();
return smp_processor_id() % num;
}
return f->arr[cpu % 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 i, j;
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:

3
net/packet/internal.h
View File

@ -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;

1
tools/testing/selftests/Makefile
View File

@ -6,6 +6,7 @@ TARGETS += cpu-hotplug
TARGETS += memory-hotplug
TARGETS += efivarfs
TARGETS += net-socket
TARGETS += net-afpacket
all:
for TARGET in $(TARGETS); do \

18
tools/testing/selftests/net-afpacket/Makefile Normal file
View File

@ -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)

326
tools/testing/selftests/net-afpacket/psock_fanout.c Normal file
View File

@ -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;
}

16
tools/testing/selftests/net-afpacket/run_afpackettests Normal file
View File

@ -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