前言
在流量分析工作中经常会使用到包捕获技术，今天结合收包流程来讨论下常见的几种包捕获技术。
图解链路层收包过程
网卡将数据包转移到内存
Step1-3
网卡驱动创建了报文描述符环形队列，并为队列中的每个描述符分配了sk_buff和内存，并将内存进行DMA映射。当网卡驱动填充了报文描述中的报文缓冲区地址后就会更新RDT寄存器的值，使之指向下一个即将填充地址信息并给网卡使用的描述符，所以网卡可以根据RDT寄存器知道自己当前可用的描述符信息。
Step4-6
此时网卡接收到数据包后，如果mac地址匹配，会将数据包写入Rx FIFO。DMA找到rx descriptor ring中下一个将要使用的descriptor，待数据包写入Rx FIFO完成后，再把Rx FIFO中的数据包复制到descriptor指定的内存中。
硬中断处理
Step7-10
复制完后，网卡启动硬中断通知CPU数据缓存区中已经有新的数据包了，以内核自带的万兆网卡驱动ixgbe为例，代码路径drivers/net/ethernet/intel/ixgbe。（本文后续涉及到的内核代码部分均为v3.10版本）。ixgbe interrupt handler入口为ixgbe_msix_clean_rings，只有一个操作调用napi_schedule。如果 NAPI processing loop尚未处于活动状态，则调用它会唤醒 NAPI processing loop。请注意，NAPI processing loop在 softirq 中执行；NAPI processing loop不会从中断处理程序中执行。NAPI 的存在专门用于收集网络数据，而无需 NIC 中断来表示数据已准备好进行处理。换句话说：NAPI 已启用，但处于关闭状态，直到第一个数据包到达时，NIC 引发 IRQ 并启动 NAPI。
static irqreturn_t ixgbe_msix_clean_rings(int irq, void *data){
struct ixgbe_q_vector *q_vector = data;
/* EIAM disabled interrupts (on this vector) for us */
if (q_vector->rx.ring || q_vector->tx.ring)
napi_schedule(&q_vector->napi);
return IRQ_HANDLED;
}
napi_schedule首先调用local_irq_save关闭中断，然后调用__napi_schedule实际执行____napi_schedule，____napi_schedule调用__raise_softirq_irqoff来触发一个 NET_RX_SOFTIRQ 软中断。如果当前未执行，将导致执行网络设备子系统初始化期间注册的 net_rx_action。
static inline void napi_schedule(struct napi_struct *n){
if (napi_schedule_prep(n))
__napi_schedule(n);
}
void __napi_schedule(struct napi_struct *n)
{
unsigned long flags;
local_irq_save(flags);//实现为宏，先保存当前终端状态到flags，再禁用中断
____napi_schedule(&__get_cpu_var(softnet_data), n);
local_irq_restore(flags);
}
static inline void ____napi_schedule(struct softnet_data *sd,
struct napi_struct *napi)
{
list_add_tail(&napi->poll_list, &sd->poll_list);
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
}
软中断处理
Step11
内核在初始化的时候，每个CPU上都会启动一个专门的ksoftirqd内核线程用于处理CPU上的软中断，接收数据包的软中断处理函数被注册为net_rx_action。一旦 softirq 代码确定一个 softirq 处于挂起状态，开始处理并执行net_rx_action，网络数据处理就开始了。该函数遍历为当前 CPU 排队的 NAPI 结构列表，使每个结构出队并对其进行操作。处理循环限制了注册的 NAPI poll函数可以消耗的工作量和执行时间，通过以下两个参数来进行限制：
-
netdev_budget 一个轮询周期中从所有接口中取出的最大数据包数量
-
netdev_budget_usecs 一个轮询周期中的最大耗时 2 jiffies
static void net_rx_action(struct softirq_action *h){
/*取得本地cpu 的softnet_data 的poll_list 链表
在里面放着所有需要轮询接收数据包的设备
*/
struct softnet_data *sd = &__get_cpu_var(softnet_data);
unsigned long time_limit = jiffies + 2;
int budget = netdev_budget;
void *have;
local_irq_disable();
//限制轮询处理包个数以及处理的时间
while (!list_empty(&sd->poll_list)) {
struct napi_struct *n;
int work, weight;
if (unlikely(budget <= 0 || time_after_eq(jiffies, time_limit)))
goto softnet_break;
......
weight = n->weight;
work = 0;
//此处poll为ixgbe_poll()
if (test_bit(NAPI_STATE_SCHED, &n->state)) {
work = n->poll(n, weight);
trace_napi_poll(n);
}
......
}
}
int ixgbe_poll(struct napi_struct *napi, int budget){
struct ixgbe_q_vector *q_vector =
container_of(napi, struct ixgbe_q_vector, napi);
struct ixgbe_adapter *adapter = q_vector->adapter;
struct ixgbe_ring *ring;
int per_ring_budget;
bool clean_complete = true;
#ifdef CONFIG_IXGBE_DCA
if (adapter->flags & IXGBE_FLAG_DCA_ENABLED)
ixgbe_update_dca(q_vector);
#endif
ixgbe_for_each_ring(ring, q_vector->tx)
clean_complete &= !!ixgbe_clean_tx_irq(q_vector, ring);
/* attempt to distribute budget to each queue fairly, but don't allow
* the budget to go below 1 because we'll exit polling */
if (q_vector->rx.count > 1)
per_ring_budget = max(budget/q_vector->rx.count, 1);
else
per_ring_budget = budget;
ixgbe_for_each_ring(ring, q_vector->rx)
clean_complete &= ixgbe_clean_rx_irq(q_vector, ring,
per_ring_budget);
/* If all work not completed, return budget and keep polling */
if (!clean_complete)
return budget;
/* all work done, exit the polling mode */
napi_complete(napi);
if (adapter->rx_itr_setting & 1)
ixgbe_set_itr(q_vector);
if (!test_bit(__IXGBE_DOWN, &adapter->state))
ixgbe_irq_enable_queues(adapter, ((u64)1 << q_vector->v_idx));
return 0;
}
ixgbe_clean_rx_irq调用ixgbe_fetch_rx_buffer根据描述符从接收队列中提取数据包，经过一些检查和处理后，调用ixgbe_rx_skb向上发送
static bool ixgbe_clean_rx_irq(struct ixgbe_q_vector *q_vector,
struct ixgbe_ring *rx_ring,
const int budget){
unsigned int total_rx_bytes = 0, total_rx_packets = 0;
#ifdef IXGBE_FCOE
struct ixgbe_adapter *adapter = q_vector->adapter;
int ddp_bytes;
unsigned int mss = 0;
#endif /* IXGBE_FCOE */
u16 cleaned_count = ixgbe_desc_unused(rx_ring);
do {
union ixgbe_adv_rx_desc *rx_desc;
struct sk_buff *skb;
......
rx_desc = IXGBE_RX_DESC(rx_ring, rx_ring->next_to_clean);
......
/* retrieve a buffer from the ring */
skb = ixgbe_fetch_rx_buffer(rx_ring, rx_desc);
......
ixgbe_rx_skb(q_vector, skb);
/* update budget accounting */
total_rx_packets++;
} while (likely(total_rx_packets < budget));
......
}
ixgbe_rx_skb判断是否支持GRO，如果支持则调用napi_gro_receive。GRO是针对报文接收方向的，是指设备链路层在接收报文处理的时候，将多个小包合并成一个大包一起上送协议栈，减少数据包在协议栈间交互的机制。使用ethtool -k eth4|grep generic-receive-offload 命令判断网卡是否启用GRO。
static void ixgbe_rx_skb(struct ixgbe_q_vector *q_vector,
struct sk_buff *skb)
{
struct ixgbe_adapter *adapter = q_vector->adapter;
if (!(adapter->flags & IXGBE_FLAG_IN_NETPOLL))
napi_gro_receive(&q_vector->napi, skb);
else
netif_rx(skb);
}
gro的处理过程略过，经过下面的调用关系，最终调用到netif_receive_skb将数据向上传递到协议层
napi_gro_receive->dev_gro_receive-> napi_gro_complete->netif_receive_skb
接下来看看数据是如何传递给协议层的
int netif_receive_skb(struct sk_buff *skb){
net_timestamp_check(netdev_tstamp_prequeue, skb);
if (skb_defer_rx_timestamp(skb))
return NET_RX_SUCCESS;
#ifdef CONFIG_RPS
if (static_key_false(&rps_needed)) {
struct rps_dev_flow voidflow, *rflow = &voidflow;
int cpu, ret;
rcu_read_lock();
cpu = get_rps_cpu(skb->dev, skb, &rflow);
if (cpu >= 0) {
ret = enqueue_to_backlog(skb, cpu, &rflow->last_qtail);
rcu_read_unlock();
return ret;
}
rcu_read_unlock();
}
#endif
return __netif_receive_skb(skb);
}
首先判断是否启用了RPS机制，RPS是早期单队列网卡上将软中断负载均衡到多个CPU Core的技术，它对数据流进行hash并分配到对应的CPU Core上，发挥多核的性能。不过现在基本都是多队列网卡，不会开启这个机制，因此走不到这里，实际上执行的是__netif_receive_skb，紧接着调用__netif_receive_skb_core。__netif_receive_skb_core 完成将数据送到协议栈这一繁重工作，在此之前，它会先检查是否插入了 packet tap（探测点），这些 tap 是抓包用的。如果有packet tap，packet 会送到那里。例如，PF_PACKET 地址族就可以插入这些抓包指令， 一般通过 libpcap 库。
static int __netif_receive_skb_core(struct sk_buff *skb, bool pfmemalloc){
......
//pcap逻辑，这里会将数据送入抓包点。tcpdump就是从这个入口获取包的
list_for_each_entry_rcu(ptype, &ptype_all, list) {
if(!ptype->dev || ptype->dev == skb->dev) {
if(pt_prev)
ret = deliver_skb(skb, pt_prev, orig_dev);
pt_prev = ptype;
}
}
......
list_for_each_entry_rcu(ptype,
&ptype_base[ntohs(type) & PTYPE_HASH_MASK], list) {
if(ptype->type == type &&
(ptype->dev == null_or_dev || ptype->dev == skb->dev ||
ptype->dev == orig_dev)) {
if(pt_prev)
ret = deliver_skb(skb, pt_prev, orig_dev);
pt_prev = ptype;
}
}
}
接着__netif_receive_skb_core取出protocol，它会从数据包中取出协议信息，然后遍历注册在这个协议上的回调函数列表。ptype_base 是一个 hash table，在协议注册小节我们提到过。ip_rcv 函数地址就是存在这个 hash table中的。
static inline int deliver_skb(struct sk_buff *skb,
struct packet_type *pt_prev,
struct net_device *orig_dev){
......
return pt_prev->func(skb, skb->dev, pt_prev, orig_dev);
}
pt_prev->func这一行就调用到了协议层注册的处理函数了。对于ip包来讲，就会进入到ip_rcv
至此，链路层的工作就完成了。上层协议栈的处理过程暂不深入，下面介绍集中常见的包捕获技术。
包捕获技术
libpcap
旧版libpcap使用传统数据包捕获机制利用PF_PACKET协议族原始套接字进行包捕获，PF_PACKET协议族是与系统TCP/IP协议栈并行的同级别模块，即从PF_PACKET协议族得到的数据包是没有经过系统TCP/IP协议栈处理的。如图2所示该机制在数据链路层增加一个旁路处理，并不干扰系统自身的网路协议栈的处理，在捕获到达网卡的数据包后绕开了传统linux协议栈处理，直接使用链路PF_PACKET协议族原始套接字方式向用户空间传递报文。反过来讲 ，用户空间可以直接调用套接字PF_PACKET从链路层驱动程序中获得数据报文的拷贝，将其从内核缓冲区拷贝至用户空间缓冲区。
应用层通过调用系统调用创建PF_PACKET socket，socket(PF_PACKET, SOCK_DGRAM, htons(ETH_P_ALL)); ETH_P_ALL表示所有的底层包都会给到PF_PACKET 模块的处理函数。在socket创建过程中创建packet_type，并挂载到全局ptype_all链表上，同时在ptype_type设置回调函数packet_rcv。在数据包到达收包处理函数__netif_receive_skb_core时，会遍历ptype_all，同时执行deliver_skb函数调用paket_type.func()，即packet_rcv函数，packet_rcv调用run_filter通过BPF过滤我们设定条件的报文，然后copy到当前的接收缓存中。最后一步应用层调用recvfrom时执行packet_recvmsg将接收缓存中的数据包copy给应用层。
传统的数据包捕获瓶颈往往在于Linux Kernel，数据流需要经过Linux Kernel，就会带来Kernel Spcae和User Space数据拷贝的消耗、系统调用的消耗、中断处理的消耗等。
libpcap-mmap
libpcap在进行高速网络流量的捕获过程非常低效，主要有两个原因：
-
在数据包捕获期间，如果数据包一到达就交付，则捕获数据包的应用程序将为每个数据包唤醒，并且可能必须对操作系统进行一次或多次调用以获取每个数据包。相反，如果数据包在到达后没有立即交付，而是在缓冲区中累积并在短暂延迟后交付，则可以减少 CPU 开销，因为每个系统调用都会交付多个数据包。到达的数据包存储在缓冲区中，只有在某些平台上才能设置缓冲区大小。如果缓冲区太小并且捕获的数据包太多，如果缓冲区在应用程序读取数据包之前填满，则可能会丢弃 pcakets。
-
数据包是在内核空间中接收的，而进程在用户空间中运行。所有的用户分析都是在用户空间中完成的。一旦内核从网络接口接收到一个数据包，它就会寻找需要该数据包的进程并将数据包传递给该进程。这种方法在捕获高速流量方面效率非常低，因为在繁重的 I/O 操作中会消耗过多的 CPU 时间。
新版本的libpcap基本都采用PACKET_MMAP机制。PACKET_MMAP（数据包内存映射技术）在内核空间中分配一块内核缓冲区，且允许用户设置自己的缓冲区大小，从而提高捕获效率。并通过在内核空间和用户空间（内存映射区域）之间使用共享缓冲区，可以最大限度地减少从内核空间到用户空间的数据包复制。读取过程只需要等待数据包写入共享环形缓冲区，减少了一次内存拷贝和一次系统调用。
PF_Ring
PF_RING是针对轮询机制的不足，在轮询机制的基础上创建的一种针对数据包捕获优化的新型套接字，它基于传入数据包被复制的循环缓冲区。工作流程如下：
-
创建PF_RING套接字时分配一个环形缓冲区，并在套接字停用时释放。不同的套接字将有一个私有的环形缓冲区。
-
如果一个PF_RING 套接字绑定到一个网卡（通过bind() 系统调用），那么该网卡将在只读模式下使用，直到该套接字被销毁。
-
每当从网卡接收到数据包时（通常通过 DMA，直接内存访问），驱动程序将数据包传递到上层（在 Linux 上，这是由 netif_receive_skb 和 netif_rx 函数实现，具体取决于是否启用轮询）。在 PF_RING 套接字的情况下，每个传入的数据包都被复制到套接字环中或在必要时丢弃。
-
接收到的绑定 PF_RING 套接字的网卡的数据包，默认情况下不会转发到上层，而是在复制到套接字环形缓冲区后会被丢弃。这种做法提高了整体性能，因为数据包不需要由上层处理，而只需要由套接字环形缓冲区处理。
-
套接字环形缓冲区通过 mmap 导出到用户空间应用程序。
-
想要访问缓冲区的用户空间应用程序需要打开文件，然后对其调用 mmap() 以获得指向环形缓冲区的指针。
-
内核将数据包复制到环形缓冲区中并向前移动写指针。用户空间应用程序对读取指针执行相同的操作。新传入的数据包会覆盖用户空间应用程序已读取的数据包。内存不会因为数据包被写/读到缓冲区来分配/释放，而是被简单地覆盖。
-
缓冲区长度和bucket大小是完全用户可配置的，并且对于所有套接字都是相同的。
-
数据包不排队进入内核网络数据结构。
-
mmap允许用户空间应用程序访问循环缓冲区，不会像套接字调用那样因系统调用而产生开销。
-
即使使用不支持设备轮询的内核，在强大的流量条件下系统也可以使用。这是因为与正常数据包处理相比，处理中断所需的时间非常有限。
-
实现数据包采样非常简单有效，因为采样的数据包不需要像传统的基于libpcap的应用程序那样被传递到上层然后被丢弃。
-
多个应用程序可以同时打开多个 PF_RING 套接字而不会相互干扰。
PF_RING支持transparent_mode 0/1/2 三种方式将裸数据放到mmap到用户态的环形缓冲区
-
transparent_mode 0
按照PACKET套接字的方式从netif_receive_skb函数中抓取数据包，这是一种和PACKET套接字兼容的方式，所不同的是数据包不再通过socket IO进入用户态，而是通过mmap()。这种方式跟新版libpcap原理相同。 -
transparent_mode 1
在这种模式下，数据包被传递到 NAPI（用于将它们发送到不支持 PF_RING 的应用程序）并直接复制到 PF_RING 以用于支持 PF_RING 的应用程序（即 PF_RING 不需要 NAPI 来接收数据包）。在这种模式下，数据包捕获会加速，因为数据包由 NIC 驱动程序复制，而无需通过通常的内核路径。 -
transparent_mode 2
驱动程序仅将数据包直接复制到 PF_RING（即 NAPI 不接收任何数据包）。这就意味着，数据包将不会进入内核，而是直接被mmap到了用户态。既然内核不需要处理网络数据了，那么CPU将被节省下来用于用户态的网络处理。
PF_Ring还提供一种更高效的方式PF_Ring ZC(零拷贝)，直接绕过内核协议栈的所有路径，也就是说直接在网卡的芯片中将数据包传输到(DMA的方式)所谓环形缓冲区，内核将看不到任何数据包。
PF_RING ZC 实现了PF_RING™ DNA（Direct NIC Access 直接网卡访问）技术。是一种映射网卡内存和寄存器到用户态的方法。
-
因此除了由网卡的网络处理单元完成DMA传输之外，没有任何额外的数据包复制，进一步节省了一次数据拷贝操作。
-
这将性能更好，因为CPU周期的仅用于操作数据包，而不是把数据包从网卡挪走。
其缺点是，只有一个应用可以在某个时间打开DMA ring（请注意，现在的网卡可以具有多个RX / TX队列，从而就可以在每个队列上同时一个应用程序），换而言之，用户态的多个应用需要彼此沟通才能分发数据包。
上图展示了Non-DNA PF_Ring和DNA PF_Ring不同的数据拷贝路径。
总结
就上文提到的几种包捕获技术来看，PF_RING ZC因为实现了DNA技术，无疑是性能最佳；PF_RING mode2 完全不使用NAPI，节省了CPU，性能次之；PF_RING mode1 部分使用NAPI，再次之；PF_RING mode0跟新版的libpcap相同使用PF_PACKET+mmap的方式，更次之；最后就是原始libpcap使用PF_PACKET套接字+socketio的方式，性能最差。
参考文档
-
https://www.ntop.org/products/packet-capture/pf_ring/
-
https://blog.csdn.net/armlinuxww/article/details/111930788
-
https://blog.packagecloud.io/monitoring-tuning-linux-networking-stack-receiving-data/#overview
-
https://docs.kernel.org/networking/packet_mmap.html
SRC活动进行中：9.16-9.30，陌陌、创新产品双倍季
原文始发于微信公众号（陌陌安全）：收包流程与包捕获技术介绍
- 左青龙
- 微信扫一扫
-
- 右白虎
- 微信扫一扫
-
评论