Commit 77f65ebd authored by Willem de Bruijn's avatar Willem de Bruijn Committed by David S. Miller

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: default avatarWillem de Bruijn <willemb@google.com>
Reviewed-by: default avatarEric Dumazet <edumazet@google.com>
Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parent b0aa73bf
...@@ -55,6 +55,8 @@ struct sockaddr_ll { ...@@ -55,6 +55,8 @@ struct sockaddr_ll {
#define PACKET_FANOUT_HASH 0 #define PACKET_FANOUT_HASH 0
#define PACKET_FANOUT_LB 1 #define PACKET_FANOUT_LB 1
#define PACKET_FANOUT_CPU 2 #define PACKET_FANOUT_CPU 2
#define PACKET_FANOUT_ROLLOVER 3
#define PACKET_FANOUT_FLAG_ROLLOVER 0x1000
#define PACKET_FANOUT_FLAG_DEFRAG 0x8000 #define PACKET_FANOUT_FLAG_DEFRAG 0x8000
struct tpacket_stats { struct tpacket_stats {
......
...@@ -181,6 +181,8 @@ static int packet_set_ring(struct sock *sk, union tpacket_req_u *req_u, ...@@ -181,6 +181,8 @@ static int packet_set_ring(struct sock *sk, union tpacket_req_u *req_u,
struct packet_sock; struct packet_sock;
static int tpacket_snd(struct packet_sock *po, struct msghdr *msg); 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, static void *packet_previous_frame(struct packet_sock *po,
struct packet_ring_buffer *rb, struct packet_ring_buffer *rb,
...@@ -973,11 +975,11 @@ static void *packet_current_rx_frame(struct packet_sock *po, ...@@ -973,11 +975,11 @@ static void *packet_current_rx_frame(struct packet_sock *po,
static void *prb_lookup_block(struct packet_sock *po, static void *prb_lookup_block(struct packet_sock *po,
struct packet_ring_buffer *rb, struct packet_ring_buffer *rb,
unsigned int previous, unsigned int idx,
int status) int status)
{ {
struct tpacket_kbdq_core *pkc = GET_PBDQC_FROM_RB(rb); 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)) if (status != BLOCK_STATUS(pbd))
return NULL; return NULL;
...@@ -1041,6 +1043,29 @@ static void packet_increment_head(struct packet_ring_buffer *buff) ...@@ -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; 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) static void packet_sock_destruct(struct sock *sk)
{ {
skb_queue_purge(&sk->sk_error_queue); skb_queue_purge(&sk->sk_error_queue);
...@@ -1066,16 +1091,16 @@ static int fanout_rr_next(struct packet_fanout *f, unsigned int num) ...@@ -1066,16 +1091,16 @@ static int fanout_rr_next(struct packet_fanout *f, unsigned int num)
return x; 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; return (((u64)skb->rxhash) * num) >> 32;
idx = ((u64)hash * num) >> 32;
return f->arr[idx];
} }
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; int cur, old;
...@@ -1083,14 +1108,40 @@ static struct sock *fanout_demux_lb(struct packet_fanout *f, struct sk_buff *skb ...@@ -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, while ((old = atomic_cmpxchg(&f->rr_cur, cur,
fanout_rr_next(f, num))) != cur) fanout_rr_next(f, num))) != cur)
cur = old; 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, 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, ...@@ -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; struct packet_fanout *f = pt->af_packet_priv;
unsigned int num = f->num_members; unsigned int num = f->num_members;
struct packet_sock *po; struct packet_sock *po;
struct sock *sk; unsigned int idx;
if (!net_eq(dev_net(dev), read_pnet(&f->net)) || if (!net_eq(dev_net(dev), read_pnet(&f->net)) ||
!num) { !num) {
...@@ -1110,23 +1161,31 @@ static int packet_rcv_fanout(struct sk_buff *skb, struct net_device *dev, ...@@ -1110,23 +1161,31 @@ static int packet_rcv_fanout(struct sk_buff *skb, struct net_device *dev,
switch (f->type) { switch (f->type) {
case PACKET_FANOUT_HASH: case PACKET_FANOUT_HASH:
default: default:
if (f->defrag) { if (fanout_has_flag(f, PACKET_FANOUT_FLAG_DEFRAG)) {
skb = ip_check_defrag(skb, IP_DEFRAG_AF_PACKET); skb = ip_check_defrag(skb, IP_DEFRAG_AF_PACKET);
if (!skb) if (!skb)
return 0; return 0;
} }
skb_get_rxhash(skb); skb_get_rxhash(skb);
sk = fanout_demux_hash(f, skb, num); idx = fanout_demux_hash(f, skb, num);
break; break;
case PACKET_FANOUT_LB: case PACKET_FANOUT_LB:
sk = fanout_demux_lb(f, skb, num); idx = fanout_demux_lb(f, skb, num);
break; break;
case PACKET_FANOUT_CPU: 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; 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); 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) ...@@ -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_sock *po = pkt_sk(sk);
struct packet_fanout *f, *match; struct packet_fanout *f, *match;
u8 type = type_flags & 0xff; u8 type = type_flags & 0xff;
u8 defrag = (type_flags & PACKET_FANOUT_FLAG_DEFRAG) ? 1 : 0; u8 flags = type_flags >> 8;
int err; int err;
switch (type) { switch (type) {
case PACKET_FANOUT_ROLLOVER:
if (type_flags & PACKET_FANOUT_FLAG_ROLLOVER)
return -EINVAL;
case PACKET_FANOUT_HASH: case PACKET_FANOUT_HASH:
case PACKET_FANOUT_LB: case PACKET_FANOUT_LB:
case PACKET_FANOUT_CPU: case PACKET_FANOUT_CPU:
...@@ -1203,7 +1265,7 @@ static int fanout_add(struct sock *sk, u16 id, u16 type_flags) ...@@ -1203,7 +1265,7 @@ static int fanout_add(struct sock *sk, u16 id, u16 type_flags)
} }
} }
err = -EINVAL; err = -EINVAL;
if (match && match->defrag != defrag) if (match && match->flags != flags)
goto out; goto out;
if (!match) { if (!match) {
err = -ENOMEM; err = -ENOMEM;
...@@ -1213,7 +1275,7 @@ static int fanout_add(struct sock *sk, u16 id, u16 type_flags) ...@@ -1213,7 +1275,7 @@ static int fanout_add(struct sock *sk, u16 id, u16 type_flags)
write_pnet(&match->net, sock_net(sk)); write_pnet(&match->net, sock_net(sk));
match->id = id; match->id = id;
match->type = type; match->type = type;
match->defrag = defrag; match->flags = flags;
atomic_set(&match->rr_cur, 0); atomic_set(&match->rr_cur, 0);
INIT_LIST_HEAD(&match->list); INIT_LIST_HEAD(&match->list);
spin_lock_init(&match->lock); spin_lock_init(&match->lock);
...@@ -3240,7 +3302,8 @@ static int packet_getsockopt(struct socket *sock, int level, int optname, ...@@ -3240,7 +3302,8 @@ static int packet_getsockopt(struct socket *sock, int level, int optname,
case PACKET_FANOUT: case PACKET_FANOUT:
val = (po->fanout ? val = (po->fanout ?
((u32)po->fanout->id | ((u32)po->fanout->id |
((u32)po->fanout->type << 16)) : ((u32)po->fanout->type << 16) |
((u32)po->fanout->flags << 24)) :
0); 0);
break; break;
case PACKET_TX_HAS_OFF: case PACKET_TX_HAS_OFF:
......
...@@ -77,10 +77,11 @@ struct packet_fanout { ...@@ -77,10 +77,11 @@ struct packet_fanout {
unsigned int num_members; unsigned int num_members;
u16 id; u16 id;
u8 type; u8 type;
u8 defrag; u8 flags;
atomic_t rr_cur; atomic_t rr_cur;
struct list_head list; struct list_head list;
struct sock *arr[PACKET_FANOUT_MAX]; struct sock *arr[PACKET_FANOUT_MAX];
int next[PACKET_FANOUT_MAX];
spinlock_t lock; spinlock_t lock;
atomic_t sk_ref; atomic_t sk_ref;
struct packet_type prot_hook ____cacheline_aligned_in_smp; struct packet_type prot_hook ____cacheline_aligned_in_smp;
......
...@@ -6,6 +6,7 @@ TARGETS += cpu-hotplug ...@@ -6,6 +6,7 @@ TARGETS += cpu-hotplug
TARGETS += memory-hotplug TARGETS += memory-hotplug
TARGETS += efivarfs TARGETS += efivarfs
TARGETS += net-socket TARGETS += net-socket
TARGETS += net-afpacket
all: all:
for TARGET in $(TARGETS); do \ for TARGET in $(TARGETS); do \
......
# 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)
/*
* 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;
}
#!/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
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment