阅读量:1
一 前言
上一篇介绍了通过AF-PACKET的V1 版本进行网络包的捕获，比较新的Linux内核是支持V3版本的，相对于前两个版本（V2和V1比较相似，V2版本的时间精度从微秒提升到纳秒。）V3版本，具有以下的提升：
CPU使用率降低约15-20%
数据包捕获率提高约20%
数据包的密度提升2倍（不知道什么意思， 如 ~2x increase in packet density）
端口聚合分析
非静态数据帧大小，可以保存整个数据包。
所以这次就学习V3版本的用法，和其他能提示AF-PACKET抓包性能的均衡策略和方法。
二 V3版本的实战
V3的版本结构每次遍历和以前的不同是按照block遍历，当然下一层再按照frame遍历。
V3的时间戳精确度到纳秒。
struct tpacket_req3 { unsigned int tp_block_size; // 每个连续内存块的最小尺寸(必须是 PAGE_SIZE * 2^n ) unsigned int tp_block_nr; // 内存块数量 unsigned int tp_frame_size; // 每个帧的大小(虽然V3中的帧长是可变的，但创建时还是会传入一个最大的允许值) unsigned int tp_frame_nr; // 帧的总个数(必须等于 每个内存块中的帧数量*内存块数量) unsigned int tp_retire_blk_tov; // 内存块的寿命(ms)，超时后即使内存块没有被数据填入也会被内核停用，0意味着不设超时 unsigned int tp_sizeof_priv; // 每个内存块中私有空间大小，0意味着不设私有空间 unsigned int tp_feature_req_word;// 标志位集合(目前就支持1个标志 TP_FT_REQ_FILL_RXHASH) } // TPACKET_V3环形缓冲区每个帧的头部结构 struct tpacket3_hdr { __u32 tp_next_offset; // 指向同一个内存块中的下一个帧 __u32 tp_sec; // 时间戳(s) __u32 tp_nsec; // 时间戳(ns) __u32 tp_snaplen; // 捕获到的帧实际长度 __u32 tp_len; // 帧的理论长度 __u32 tp_status; // 帧的状态 __u16 tp_mac; // 以太网MAC字段距离帧头的偏移量 __u16 tp_net; union { struct tpacket_hdr_variant1 hv1; // 包含vlan信息的子结构 }; __u8 tp_padding[8]; }下面是内核文档中的收包例子，代码如下：
/* Written from scratch, but kernel-to-user space API usage * dissected from lolpcap: * Copyright 2011, Chetan Loke <loke.chetan@gmail.com> * License: GPL, version 2.0 */ #include <stdio.h> #include <stdlib.h> #include <stdint.h> #include <string.h> #include <assert.h> #include <net/if.h> #include <arpa/inet.h> #include <netdb.h> #include <poll.h> #include <unistd.h> #include <signal.h> #include <inttypes.h> #include <sys/socket.h> #include <sys/mman.h> #include <linux/if_packet.h> #include <linux/if_ether.h> #include <linux/ip.h> #ifndef likely # define likely(x) __builtin_expect(!!(x), 1) #endif #ifndef unlikely # define unlikely(x) __builtin_expect(!!(x), 0) #endif struct block_desc { uint32_t version; uint32_t offset_to_priv; struct tpacket_hdr_v1 h1; }; struct ring { struct iovec *rd; uint8_t *map; struct tpacket_req3 req; }; static unsigned long packets_total = 0, bytes_total = 0; static sig_atomic_t sigint = 0; static void sighandler(int num) { sigint = 1; } static int setup_socket(struct ring *ring, char *netdev) { int err, i, fd, v = TPACKET_V3; struct sockaddr_ll ll; unsigned int blocksiz = 1 << 22, framesiz = 1 << 11; unsigned int blocknum = 64; fd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL)); if (fd < 0) { perror("socket"); exit(1); } err = setsockopt(fd, SOL_PACKET, PACKET_VERSION, &v, sizeof(v)); if (err < 0) { perror("setsockopt"); exit(1); } memset(&ring->req, 0, sizeof(ring->req)); ring->req.tp_block_size = blocksiz; ring->req.tp_frame_size = framesiz; ring->req.tp_block_nr = blocknum; ring->req.tp_frame_nr = (blocksiz * blocknum) / framesiz; ring->req.tp_retire_blk_tov = 60; ring->req.tp_feature_req_word = TP_FT_REQ_FILL_RXHASH; err = setsockopt(fd, SOL_PACKET, PACKET_RX_RING, &ring->req, sizeof(ring->req)); if (err < 0) { perror("setsockopt"); exit(1); } ring->map = mmap(NULL, ring->req.tp_block_size * ring->req.tp_block_nr, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_LOCKED, fd, 0); if (ring->map == MAP_FAILED) { perror("mmap"); exit(1); } ring->rd = malloc(ring->req.tp_block_nr * sizeof(*ring->rd)); assert(ring->rd); for (i = 0; i < ring->req.tp_block_nr; ++i) { ring->rd[i].iov_base = ring->map + (i * ring->req.tp_block_size); ring->rd[i].iov_len = ring->req.tp_block_size; } memset(&ll, 0, sizeof(ll)); ll.sll_family = PF_PACKET; ll.sll_protocol = htons(ETH_P_ALL); ll.sll_ifindex = if_nametoindex(netdev); ll.sll_hatype = 0; ll.sll_pkttype = 0; ll.sll_halen = 0; err = bind(fd, (struct sockaddr *) &ll, sizeof(ll)); if (err < 0) { perror("bind"); exit(1); } return fd; } static void display(struct tpacket3_hdr *ppd) { struct ethhdr *eth = (struct ethhdr *) ((uint8_t *) ppd + ppd->tp_mac); struct iphdr *ip = (struct iphdr *) ((uint8_t *) eth + ETH_HLEN); if (eth->h_proto == htons(ETH_P_IP)) { struct sockaddr_in ss, sd; char sbuff[NI_MAXHOST], dbuff[NI_MAXHOST]; memset(&ss, 0, sizeof(ss)); ss.sin_family = PF_INET; ss.sin_addr.s_addr = ip->saddr; getnameinfo((struct sockaddr *) &ss, sizeof(ss), sbuff, sizeof(sbuff), NULL, 0, NI_NUMERICHOST); memset(&sd, 0, sizeof(sd)); sd.sin_family = PF_INET; sd.sin_addr.s_addr = ip->daddr; getnameinfo((struct sockaddr *) &sd, sizeof(sd), dbuff, sizeof(dbuff), NULL, 0, NI_NUMERICHOST); printf("%s -> %s, ", sbuff, dbuff); } printf("rxhash: 0x%x\n", ppd->hv1.tp_rxhash); } static void walk_block(struct block_desc *pbd, const int block_num) { int num_pkts = pbd->h1.num_pkts, i; unsigned long bytes = 0; struct tpacket3_hdr *ppd; ppd = (struct tpacket3_hdr *) ((uint8_t *) pbd + pbd->h1.offset_to_first_pkt); for (i = 0; i < num_pkts; ++i) { bytes += ppd->tp_snaplen; display(ppd); ppd = (struct tpacket3_hdr *) ((uint8_t *) ppd + ppd->tp_next_offset); } packets_total += num_pkts; bytes_total += bytes; } static void flush_block(struct block_desc *pbd) { pbd->h1.block_status = TP_STATUS_KERNEL; } static void teardown_socket(struct ring *ring, int fd) { munmap(ring->map, ring->req.tp_block_size * ring->req.tp_block_nr); free(ring->rd); close(fd); } int main(int argc, char **argp) { int fd, err; socklen_t len; struct ring ring; struct pollfd pfd; unsigned int block_num = 0, blocks = 64; struct block_desc *pbd; struct tpacket_stats_v3 stats; if (argc != 2) { fprintf(stderr, "Usage: %s INTERFACE\n", argp[0]); return EXIT_FAILURE; } signal(SIGINT, sighandler); memset(&ring, 0, sizeof(ring)); fd = setup_socket(&ring, argp[argc - 1]); assert(fd > 0); memset(&pfd, 0, sizeof(pfd)); pfd.fd = fd; pfd.events = POLLIN | POLLERR; pfd.revents = 0; while (likely(!sigint)) { pbd = (struct block_desc *) ring.rd[block_num].iov_base; if ((pbd->h1.block_status & TP_STATUS_USER) == 0) { poll(&pfd, 1, -1); continue; } walk_block(pbd, block_num); flush_block(pbd); block_num = (block_num + 1) % blocks; } len = sizeof(stats); err = getsockopt(fd, SOL_PACKET, PACKET_STATISTICS, &stats, &len); if (err < 0) { perror("getsockopt"); exit(1); } fflush(stdout); printf("\nReceived %u packets, %lu bytes, %u dropped, freeze_q_cnt: %u\n", stats.tp_packets, bytes_total, stats.tp_drops, stats.tp_freeze_q_cnt); teardown_socket(&ring, fd); return 0; }代码说明：
使用的tpacket_req3 时候会有两个以前没有的变量赋值如下：
ring->req.tp_retire_blk_tov = 60; ring->req.tp_feature_req_word = TP_FT_REQ_FILL_RXHASH;tp_retrie_blk_tov 即超时值，单位是毫秒。当这个超时被触发时，内核会将该块的状态从 TP_STATUS_USER 改为 TP_STATUS_KERNEL，即超时触发后，会将缓存区释放给内核用来装数据了。
tp_feature_req_word 即功能值，tp_feature_req_word 被设置为 TP_FT_REQ_FILL_RXHASH。这个标志请求内核在每个数据包的头部填充接收哈希（RX hash）值。接收哈希通常用于负载均衡和流量分类等场景，它可以提供一种快速的方式来决定数据包应该如何被处理或路由。
v3版本来说，数据包帧的最小大小，不能低于这个大小。
mmap申请内核和用户空间共享的内存，注意设置MAP_LOCKED标识的意思，这个内存是不可以交换到磁盘上的，会被锁在物理内存中。
如果需要让内核设置包的时间，可以通过下面代码设置：
int req = SOF_TIMESTAMPING_RAW_HARDWARE; setsockopt(fd, SOL_PACKET, PACKET_TIMESTAMP, (void *) &req, sizeof(req));转换ip代码：
struct sockaddr_in ss, sd; char sbuff[NI_MAXHOST], dbuff[NI_MAXHOST]; memset(&ss, 0, sizeof(ss)); ss.sin_family = PF_INET; ss.sin_addr.s_addr = ip->saddr; getnameinfo((struct sockaddr *) &ss, sizeof(ss),sbuff, sizeof(sbuff), NULL, 0, NI_NUMERICHOST);这段代码是将获取的包的ip地址，其中：NI_NUMERICHOST: 直接返回主机的IP地址，而不是尝试查找主机名。
主循环：
while (likely(!sigint)) { pbd = (struct block_desc *) ring.rd[block_num].iov_base; if ((pbd->h1.block_status & TP_STATUS_USER) == 0) { poll(&pfd, 1, -1); continue; } walk_block(pbd, block_num); flush_block(pbd); block_num = (block_num + 1) % blocks; }在没有中断的情况下，获取block头部信息，根据头的block状态，如果不是TP_STATUS_USER情况下，继续进入poll等待。如果已经有数据了，则走walk循环，然后更改block状态，block指针后移，做循环队列，这个和原来v1版本的frame做循环。
代码如下：
// 调整block执行第一个packet包、下一个包是通过tp_next_offset 来指向的。 ppd = (struct tpacket3_hdr *) ((uint8_t *) pbd + pbd->h1.offset_to_first_pkt); for (i = 0; i < num_pkts; ++i) { bytes += ppd->tp_snaplen; display(ppd); ppd = (struct tpacket3_hdr *) ((uint8_t *) ppd + ppd->tp_next_offset); }代码里面数据结构示意图如下：
image.png
三 SOCKET的fanout
虽然V3版本相当于前面的版本来说，性能有所提升，但是在现在的多核环境下，如果只简单的用V3版本，仍然，不能达到捕获超大流量数据包情况下而不丢包。
AF-PACKET支持多核抓包，从内核3.1版本开始，AF-PACKET 可以将多个不同进程或线程的socket加入到一个fanout组中，并行一同抓包。内核会利用一定的算法对packet进行分组进入不同socket中，从而让包在多个线程或进程中进行处理，达到负载均衡的目的。
多socket收包
一个fanout组中最多可以支持65536个socket，通过setsockopt系统调用并指定PACKET_FANOUT选项来加入扇出组，当关闭一个组内socket的时候，就会从扇出组内扇出，如果组内的所有socket都退出了，组也会被内核销毁。这里面的fanout翻译成扇出组，想想挺形象，像扇子一样，扇子的头是一个，打开后分散了，扇出组也是这个目的，上图中NIC是有多队列的，所以本身就是分散进入的数据。
扇出有不同的算法，说明如下：
PACKET_FANOUT_HASH: 默认的扇出算法，通过计算ip地址和可选的tcp端口的哈希函数，将包发到不同的socket，且能保持同一个流的packet发到同一个socket，便于后续线程继续处理。
PACKET_FANOUT_LB: 简单的采用轮训的负载均衡算法，但是不能保证同一个流发送到同一个socket。
PACKET_FANOUT_RND :模式通过使用伪随机数生成器来选择目标套接字。同样，这种模式只允许进行无状态处理。
PACKET_FANOUT_CPU :模式根据接收到数据包的 CPU 来选择数据包套接字。
PACKET_FANOUT_ROLLOVER: 模式持续将所有数据发送到一个套接字，直到该套接字出现拥塞。然后，它转移到组内的下一个套接字，直到该套接字也耗尽，依此类推。
PACKET_FANOUT_QM : 模式选择与接收到数据包的硬件队列编号相匹配的数据包套接字。即RSS队列和socket对应，这种默认如果设置了RSS的均衡算法为对称哈希算法，则也能保证一个流的所有packet发送到同一个socket中去。Linux 3.14以后可用，以前版本不支持。
看下内核测试代码里面的例子如下：
#include <stddef.h> #include <stdlib.h> #include <stdio.h> #include <string.h> #include <sys/types.h> #include <sys/wait.h> #include <sys/socket.h> #include <sys/ioctl.h> #include <unistd.h> #include <linux/if_ether.h> #include <linux/if_packet.h> #include <net/if.h> static const char *device_name; static int fanout_type; static int fanout_id; #ifndef PACKET_FANOUT # define PACKET_FANOUT 18 # define PACKET_FANOUT_HASH 0 # define PACKET_FANOUT_LB 1 #endif static int setup_socket(void) { int err, fd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_IP)); struct sockaddr_ll ll; struct ifreq ifr; int fanout_arg; if (fd < 0) { perror("socket"); return EXIT_FAILURE; } memset(&ifr, 0, sizeof(ifr)); strcpy(ifr.ifr_name, device_name); err = ioctl(fd, SIOCGIFINDEX, &ifr); if (err < 0) { perror("SIOCGIFINDEX"); return EXIT_FAILURE; } memset(&ll, 0, sizeof(ll)); ll.sll_family = AF_PACKET; ll.sll_ifindex = ifr.ifr_ifindex; err = bind(fd, (struct sockaddr *) &ll, sizeof(ll)); if (err < 0) { perror("bind"); return EXIT_FAILURE; } fanout_arg = (fanout_id | (fanout_type << 16)); err = setsockopt(fd, SOL_PACKET, PACKET_FANOUT, &fanout_arg, sizeof(fanout_arg)); if (err) { perror("setsockopt"); return EXIT_FAILURE; } return fd; } static void fanout_thread(void) { int fd = setup_socket(); int limit = 10000; if (fd < 0) exit(fd); while (limit-- > 0) { char buf[1600]; int err; err = read(fd, buf, sizeof(buf)); if (err < 0) { perror("read"); exit(EXIT_FAILURE); } if ((limit % 10) == 0) fprintf(stdout, "(%d) \n", getpid()); } fprintf(stdout, "%d: Received 10000 packets\n", getpid()); close(fd); exit(0); } int main(int argc, char **argp) { int fd, err; int i; if (argc != 3) { fprintf(stderr, "Usage: %s INTERFACE {hash|lb}\n", argp[0]); return EXIT_FAILURE; } if (!strcmp(argp[2], "hash")) fanout_type = PACKET_FANOUT_HASH; else if (!strcmp(argp[2], "lb")) fanout_type = PACKET_FANOUT_LB; else { fprintf(stderr, "Unknown fanout type [%s]\n", argp[2]); exit(EXIT_FAILURE); } device_name = argp[1]; fanout_id = getpid() & 0xffff; for (i = 0; i < 4; i++) { pid_t pid = fork(); switch (pid) { case 0: fanout_thread(); case -1: perror("fork"); exit(EXIT_FAILURE); } } for (i = 0; i < 4; i++) { int status; wait(&status); } return 0; }代码采用简单的方式演示了fanout代码，关键代码在于：
// fanout_arg 高16位存的是fanout的类型，低16位存的是fanout的id，即是fanout的组id。 fanout_arg = (fanout_id | (fanout_type << 16)); err = setsockopt(fd, SOL_PACKET, PACKET_FANOUT, &fanout_arg, sizeof(fanout_arg));四 参考
https://www.kernel.org/doc/Documentation/networking/packet_mmap.txt