PF_RING实现分析（1）

2020-05-22 16:36:37

版权所有，转载请注明出处

独孤九贱

内核版本：Linux 2.6.30.9

PF_RING版本：4.1.0

近看了一个PF_RING的实现，看了个大概，发上来大家讨论讨论，共同学习。

一、什么是PF_RING
PF_RING是一个第三方的内核数据包捕获接口，类似于libpcap，它的官方网址是：http://www.ntop.org/PF_RING.html

二、为什么需要PF_RING
一切为了效率，按照其官方网站上的测试数据，在Linux平台之上，其效率至少高于libpcap 50% - 60%，甚至是一倍。更好的是，PF_RING提供了一个修改版本的libpcap，使之建立在PF_RING接口之上。这样，原来使用libpcap的程序，就可以自然过渡了。

三、声明
1、这不是“零拷贝”，研究“零拷贝”的估计要失望，如果继续看下去的话；
2、这不是包截获接口，如果需要拦截、修改内核数据包，请转向Netfilter；
3、本文只分析了PF_RING基础的部份。关于DNA、TNAPI，BPF等内容不包含在内。

四、源码的获取
svn co https://svn.ntop.org/svn/ntop/trunk/PF_RING/

近好像全流行svn了。

五、编译和使用
接口分为两部份，一个是内核模块，一个是用户态的库
cd my_pf_ring_goes_here
cd kernel
make
sudo insmod ./pf_ring.ko
cd ../userland
make

在源码目录中，关于用户态的库有使用的现成的例子，很容易依葫芦画瓢。后文也会提到用户态库的实现的简单分析，可以两相比照，很容易上手。而且源码目录中有一个PDF文档，有详细的API介绍，建议使用前阅读。

六、实现分析初步
1、核心思路
A、在内核队列层注册Hook，获取数据帧。
B、在内核创建一个环形队列(这也是叫RING的原因)，用于存储数据，并使用mmap映射到用户空间。这样，避免用户态的系统调用，也是提高性能的关键所在。
C、创建了一个新的套接字类型PF_RING，用户态通过它与内核通信。

2、模块初始化
模块源码只有一个文件，在目录树kernel/pf_ring.c，嗯，还有一个头文件，在kernel/linux下

static int __init ring_init(void)
{
  int i, rc;

  printk("[PF_RING] Welcome to PF_RING %s ($Revision: 4012 $)\n"
         "(C) 2004-09 L.Deri <[email]deri@ntop.org[/email]>\n", RING_VERSION);
        
        //注册名为PF_RING的新协议
  if((rc = proto_register(&ring_proto, )) != )
    return(rc);

ring_proto的定义为

#if(LINUX_VERSION_CODE > KERNEL_VERSION(2,6,11))
static struct proto ring_proto = {
  .name = "PF_RING",
  .owner = THIS_MODULE,
  .obj_size = sizeof(struct ring_sock),
};
#endif

初始化四个链表，它们的作用，后文会分析到：

//初始化四个链表
  INIT_LIST_HEAD(&ring_table);                                        /* List of all ring sockets. */
  INIT_LIST_HEAD(&ring_cluster_list);                        /* List of all clusters */
  INIT_LIST_HEAD(&ring_aware_device_list);                /* List of all devices on which PF_RING has been registered */
  INIT_LIST_HEAD(&ring_dna_devices_list);                /* List of all dna (direct nic access) devices */

device_ring_list是一个指针数组，它的每一个元素对应一个网络设备，后文也会分析它的使用：

/*
  For each device, pf_ring keeps a list of the number of
  available ring socket slots. So that a caller knows in advance whether
  there are slots available (for rings bound to such device)
  that can potentially host the packet
*/
  for (i = ; i < MAX_NUM_DEVICES; i++)
    INIT_LIST_HEAD(&device_ring_list[i]);

//为新协议注册sock
  sock_register(&ring_family_ops);

static struct net_proto_family ring_family_ops = {
  .family = PF_RING,
  .create = ring_create,
  .owner = THIS_MODULE,
};

这样，当用户空间创建PF_RING时，例如，

fd = socket(PF_RING, SOCK_RAW, htons(ETH_P_ALL));

ring_create将会被调用

//注册通知链表  
  register_netdevice_notifier(&ring_netdev_notifier);

  /* 工作模式语法检查 */
  if(transparent_mode > driver2pf_ring_non_transparent)
    transparent_mode = standard_linux_path;

PF_RING一共有三种工作模式：

typedef enum {
  standard_linux_path = , /* Business as usual */
  driver2pf_ring_transparent = 1, /* Packets are still delivered to the kernel */
  driver2pf_ring_non_transparent = 2 /* Packets not delivered to the kernel */
} direct2pf_ring;

种简单，这里仅分析种

//输出工作信息参数
  printk("[PF_RING] Ring slots       %d\n", num_slots);
  printk("[PF_RING] Slot version     %d\n",
         RING_FLOWSLOT_VERSION);
  printk("[PF_RING] Capture TX       %s\n",
         enable_tx_capture ? "Yes [RX+TX]" : "No [RX only]");
  printk("[PF_RING] Transparent Mode %d\n",
         transparent_mode);
  printk("[PF_RING] IP Defragment    %s\n",
         enable_ip_defrag ? "Yes" : "No");
  printk("[PF_RING] Initialized correctly\n");

num_slots为槽位总数，系统采用数组来实现双向环形队列，它也就代表数组的大元素。
版本信息：不用多说了。不过我的版本在2.6.18及以下都没有编译成功，后来使用2.6.30.9搞定之。
enable_tx_capture：是否启用发送时的数据捕获，对于大多数应用而言，都是在接收时处理。
enable_ip_defrag：为用户提供一个接口，是否在捕获重组IP分片。

//创建/proc目录
  ring_proc_init();
  
  //注册设备句柄
  register_device_handler();

  pfring_enabled = 1;                //工作标志
  return ;
}

register_device_handler注册了一个协议，用于数据包的获取：

/* Protocol hook */
static struct packet_type prot_hook;

void register_device_handler(void) {
        //只有在种模式下，才用这种方式接收数据
  if(transparent_mode != standard_linux_path) return;

  prot_hook.func = packet_rcv;
  prot_hook.type = htons(ETH_P_ALL);
  dev_add_pack(&prot_hook);
}

void register_device_handler(void) {
  if(transparent_mode != standard_linux_path) return;

  prot_hook.func = packet_rcv;
  prot_hook.type = htons(ETH_P_ALL);
  dev_add_pack(&prot_hook);
}

2、创建套接字
Linux的套按字的内核接口，使用了两个重要的数据结构：
struct socket和struct sock，这本来并没有什么，不过令人常常迷惑的是，前者常常被缩写为sock，即：
struct socket *sock;
这样，“sock”就容易造成混淆了。还好，后者常常被缩写为sk……
我这里写sock指前者，sk指后者，如果不小心写混了，请参考上下文区分。

关于这两个结构的含义，使用等等，可以参考相关资料以获取详细信息，如《Linux情景分析》。我的个人网站www.skynet.org.cn上也分析了Linux socket的实现。可以参考。这里关于socket的进一步信息，就不详细分析了。

这里的创建套接字，内核已经在系统调用过程中，准备好了sock，主要就是分析sk，并为sk指定一系列的操作函数，如bind、mmap、poll等等。
如前所述，套接字的创建，是通过调用ring_create函数来完成的：

static int ring_create(
#if(LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,24))
                       struct net *net,
#endif
                       struct socket *sock, int protocol)
{
  struct sock *sk;
  struct ring_opt *pfr;
  int err;

#if defined(RING_DEBUG)
  printk("[PF_RING] ring_create()\n");
#endif

  /* 权限验证 ? */
  if(!capable(CAP_NET_ADMIN))
    return -EPERM;

  //协议簇验证
  if(sock->type != SOCK_RAW)
    return -ESOCKTNOSUPPORT;

  //协议验证
  if(protocol != htons(ETH_P_ALL))
    return -EPROTONOSUPPORT;

  err = -ENOMEM;

  // 分配sk
  // options are.
#if(LINUX_VERSION_CODE <= KERNEL_VERSION(2,6,11))
  sk = sk_alloc(PF_RING, GFP_KERNEL, 1, NULL);
#else
#if(LINUX_VERSION_CODE < KERNEL_VERSION(2,6,24))
    // BD: API changed in 2.6.12, ref:
    // [url]http://svn.clkao.org/svnweb/linux/revision/?rev=28201[/url]
    sk = sk_alloc(PF_RING, GFP_ATOMIC, &ring_proto, 1);
#else
    sk = sk_alloc(net, PF_INET, GFP_KERNEL, &ring_proto);
#endif
#endif

  //分配失败
  if(sk == NULL)
    goto out;
  
  //这里很重要，设定sock的ops，即对应用户态的bind、connect诸如此类操作的动作
  sock->ops = &ring_ops;

  //初始化sock结构(即sk)各成员，并设定与套接字socket(即sock)的关联
  sock_init_data(sock, sk);
#if(LINUX_VERSION_CODE <= KERNEL_VERSION(2,6,11))
  sk_set_owner(sk, THIS_MODULE);
#endif

  err = -ENOMEM;

#define ring_sk_datatype(__sk) ((struct ring_opt *)__sk)
#define ring_sk(__sk) ((__sk)->sk_protinfo)

//作者喜欢用小写的宏
//这里分配一个struct ring_opt结构，这个结构比较重要，其记录了ring的选项信息。
在sk中，使用sk_protinfo成员指向之，这样就建立了sock->sk->ring_opt的关联。可以通过套接字很容易获取Ring的信息。
  ring_sk(sk) = ring_sk_datatype(kmalloc(sizeof(*pfr), GFP_KERNEL));
  //分配失败
  if(!(pfr = ring_sk(sk))) {
    sk_free(sk);
    goto out;
  }

  //初始化各成员
  memset(pfr, 0, sizeof(*pfr));
//激活标志
  pfr->ring_active = 0;        /* We activate as soon as somebody waits for packets */
  //通道ID
  pfr->channel_id = RING_ANY_CHANNEL;
  //RING的每个槽位的桶的大小，其用来存储捕获的数据帧，这个值，用户态也可以使用setsocketopt来调整
  pfr->bucket_len = DEFAULT_BUCKET_LEN;
//过滤器hash桶
  pfr->handle_hash_rule = handle_filtering_hash_bucket;
  //初始化等待队列
  init_waitqueue_head(&pfr->ring_slots_waitqueue);
  //初始化RING的锁
  rwlock_init(&pfr->ring_index_lock);
  rwlock_init(&pfr->ring_rules_lock);
  //初始化使用计数器
  atomic_set(&pfr->num_ring_users, 0);
  INIT_LIST_HEAD(&pfr->rules);
  //设定协议簇
  sk->sk_family = PF_RING;
//设定sk的destuct函数
  sk->sk_destruct = ring_sock_destruct;

  //sk入队
  ring_insert(sk);

#if defined(RING_DEBUG)
  printk("[PF_RING] ring_create() - created\n");
#endif

  return(0);
out:
  return err;

在模块初始化中，初始化过四个链表。其中一个是ring_table，ring_insert将刚刚创建的套接字插入其中。其封装引进了一个struct ring_element 结构：

/*
* ring_insert()
*
* store the sk in a new element and add it
* to the head of the list.
*/
static inline void ring_insert(struct sock *sk)
{
struct ring_element *next;
struct ring_opt *pfr;
#if defined(RING_DEBUG)
printk("[PF_RING] ring_insert()\n");
#endif
//分配element
next = kmalloc(sizeof(struct ring_element), GFP_ATOMIC);
if(next != NULL) {
//记录sk
next->sk = sk;
write_lock_bh(&ring_mgmt_lock);
//入队
list_add(&next->list, &ring_table);
write_unlock_bh(&ring_mgmt_lock);
} else {
if(net_ratelimit())
printk("[PF_RING] net_ratelimit() failure\n");
}
//累计使用计数器
ring_table_size++;
//ring_proc_add(ring_sk(sk));
//记录进程PID
pfr = (struct ring_opt *)ring_sk(sk);
pfr->ring_pid = current->pid;
}

复制代码

3 、分配队列空间
用户态在创建了套接字后，接下来就调用bind函数，绑定套接字，而PF_RING实际做的就是为RING分配相应的空间。也就是说，一个套接字，都有一个与之对应的RING。这样，有多个进程同时使用PF_RING，也没有问题:

sa.sa_family = PF_RING;
snprintf(sa.sa_data, sizeof(sa.sa_data), "%s", device_name);
rc = bind(ring->fd, (struct sockaddr *)&sa, sizeof(sa));

复制代码

因为前一步创建套接字时，为sk指定了其ops：

static struct proto_ops ring_ops = {
.family = PF_RING,
.owner = THIS_MODULE,
/* Operations that make no sense on ring sockets. */
.connect = sock_no_connect,
.socketpair = sock_no_socketpair,
.accept = sock_no_accept,
.getname = sock_no_getname,
.listen = sock_no_listen,
.shutdown = sock_no_shutdown,
.sendpage = sock_no_sendpage,
.sendmsg = sock_no_sendmsg,
/* Now the operations that really occur. */
.release = ring_release,
.bind = ring_bind,
.mmap = ring_mmap,
.poll = ring_poll,
.setsockopt = ring_setsockopt,
.getsockopt = ring_getsockopt,
.ioctl = ring_ioctl,
.recvmsg = ring_recvmsg,
};

复制代码

这样，当bing系统调用触发时，ring_bind函数将被调用：

* Bind to a device */
static int ring_bind(struct socket *sock, struct sockaddr *sa, int addr_len)
{
struct sock *sk = sock->sk;
struct net_device *dev = NULL;
#if defined(RING_DEBUG)
printk("[PF_RING] ring_bind() called\n");
#endif
/*
* Check legality
*/
if(addr_len != sizeof(struct sockaddr))
return -EINVAL;
if(sa->sa_family != PF_RING)
return -EINVAL;
if(sa->sa_data == NULL)
return -EINVAL;
/* Safety check: add trailing zero if missing */
sa->sa_data[sizeof(sa->sa_data) - 1] = '\0';
#if defined(RING_DEBUG)
printk("[PF_RING] searching device %s\n", sa->sa_data);
#endif
if((dev = __dev_get_by_name(
#if(LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,24))
&init_net,
#endif
sa->sa_data)) == NULL) {
#if defined(RING_DEBUG)
printk("[PF_RING] search failed\n");
#endif
return(-EINVAL);
} else
return(packet_ring_bind(sk, dev));
}

复制代码

在做了一些必要的语法检查后，函数转向packet_ring_bind：

/*
* We create a ring for this socket and bind it to the specified device
*/
static int packet_ring_bind(struct sock *sk, struct net_device *dev)
{
u_int the_slot_len;
u_int32_t tot_mem;
struct ring_opt *pfr = ring_sk(sk);
// struct page *page, *page_end;
if(!dev)
return(-1);
#if defined(RING_DEBUG)
printk("[PF_RING] packet_ring_bind(%s) called\n", dev->name);
#endif
/* **********************************************
* *************************************
* * *
* * FlowSlotInfo *
* * *
* ************************************* <-+
* * FlowSlot * |
* ************************************* |
* * FlowSlot * |
* ************************************* +- num_slots
* * FlowSlot * |
* ************************************* |
* * FlowSlot * |
* ************************************* <-+
*
* ********************************************** */
//计算每一个槽位所需的内存空间
the_slot_len = sizeof(u_char) /* flowSlot.slot_state */
#ifdef RING_MAGIC
+ sizeof(u_char)
#endif
+ sizeof(struct pfring_pkthdr)
+ pfr->bucket_len /* flowSlot.bucket */ ;
/*
对于槽位空间的计算，有意思的是
typedef struct flowSlot {
#ifdef RING_MAGIC
u_char magic; /* It must alwasy be zero */
#endif
u_char slot_state; /* 0=empty, 1=full */
u_char bucket; /* bucket[bucketLen] */
} FlowSlot;
对照结构定义和上面的计算公式：
1、作者好像把magic和slog_state的顺序给搞反了，不过还好，它们都是u_char，对结果不影响
2、bucket，桶的大小，这个桶就是拿来装要捕获的数据包了，虽然它在结构中定义是一个成员，事实上，
它由两个部份组成，一个是包的首部信息，这个结构的定义同libpcap很接近。另一个才是包的空间。
*/
//总共的环形队列内存所需空间，包含一个队列控制信息FlowSlotInfo和若干个(由变量num_slots决定)槽位空间
tot_mem = sizeof(FlowSlotInfo) + num_slots * the_slot_len;
//确保按整数页分配，mmap也要求这样
if(tot_mem % PAGE_SIZE)
tot_mem += PAGE_SIZE - (tot_mem % PAGE_SIZE);
//分配内存空间
pfr->ring_memory = rvmalloc(tot_mem);
if(pfr->ring_memory != NULL) {
printk("[PF_RING] successfully allocated %lu bytes at 0x%08lx\n",
(unsigned long)tot_mem, (unsigned long)pfr->ring_memory);
} else {
printk("[PF_RING] ERROR: not enough memory for ring\n");
return(-1);
}
// memset(pfr->ring_memory, 0, tot_mem); // rvmalloc does the memset already
//初始化各成员
//内存指定，因为分配的内存开始部份是sizeof(FlowSlotInfo)，所以可以做这样的强制转换，很容易互相取值
pfr->slots_info = (FlowSlotInfo *) pfr->ring_memory;
//跳过控制信息，指向槽位指针.事实上，它就是一个一维数组了，可以计算出合适的索引值，取到数组(RING)中的任意槽位值
pfr->ring_slots = (char *)(pfr->ring_memory + sizeof(FlowSlotInfo));
//版本信息
pfr->slots_info->version = RING_FLOWSLOT_VERSION;
//登记单个槽的大小
pfr->slots_info->slot_len = the_slot_len;
//登记bucket大小，从前面特别注释的bucket的分配看，bucket_len这个大小并不代表bucket成员的实际大小——它不包含struct pfring_pkthdr
pfr->slots_info->data_len = pfr->bucket_len;
//登记实际分配到的槽位数量，这里不用num_slots，难道是怕rvmalloc偷吃？
pfr->slots_info->tot_slots =
(tot_mem - sizeof(FlowSlotInfo)) / the_slot_len;
//登记实际分配的内存总数
pfr->slots_info->tot_mem = tot_mem;
//采样速率？？
pfr->slots_info->sample_rate = 1;
printk("[PF_RING] allocated %d slots [slot_len=%d][tot_mem=%u]\n",
pfr->slots_info->tot_slots, pfr->slots_info->slot_len,
pfr->slots_info->tot_mem);
#ifdef RING_MAGIC
{
int i;
for (i = 0; i < pfr->slots_info->tot_slots; i++) {
unsigned long idx = i * pfr->slots_info->slot_len;
FlowSlot *slot = (FlowSlot *) & pfr->ring_slots[idx];
slot->magic = RING_MAGIC_VALUE;
slot->slot_state = 0;
}
}
#endif
//这些控制变量可以在环的入队操作中看到它们的作用
pfr->sample_rate = 1; /* No sampling */
pfr->insert_page_id = 1, pfr->insert_slot_id = 0;
pfr->rules_default_accept_policy = 1, pfr->num_filtering_rules = 0;
ring_proc_add(ring_sk(sk), dev);
//记录与之相应的设备信息，例如，如果在eth0上打开了5 个PF_RING, bind
//被调用5次，分配了5个环形队列空间。eth0上随之分配5个elem，它们指向与
//之对应的ring，然后根据设备索引号民全部加入至了device_ring_list
//当有数据报文从指定接口进入时，可以很容易地在device_ring_list中找到相应的设备
//然后再遍历链表，再找到与之相应的ring
if(dev->ifindex < MAX_NUM_DEVICES) {
device_ring_list_element *elem;
/* printk("[PF_RING] Adding ring to device index %d\n", dev->ifindex); */
elem = kmalloc(sizeof(device_ring_list_element), GFP_ATOMIC);
if(elem != NULL) {
elem->the_ring = pfr;
INIT_LIST_HEAD(&elem->list);
write_lock(&ring_list_lock);
list_add(&elem->list, &device_ring_list[dev->ifindex]);
write_unlock(&ring_list_lock);
/* printk("[PF_RING] Added ring to device index %d\n", dev->ifindex); */
}
}
/*
IMPORTANT
Leave this statement here as last one. In fact when
the ring_netdev != NULL the socket is ready to be used.
*/
pfr->ring_netdev = dev;
return(0);
}

复制代码

这个函数中，重要的三点：
1、整个空间的详细构成，作者画了一个简单的草图，清晰明了。
2、如果取得某个槽位。
3、device_ring_list链表的使用。

一些作者有详细注释的地方，我就不再重重了。

这一步进行完了后，就有一块内存了（系统将其看成一个数组），用来存储捕获的数据帧。接下来要做的事情。就是把它映射到用户态。

文章来源CU社区：PF_RING实现分析

分享好友

分享这个小栈给你的朋友们，一起进步吧。

内核源码

创建时间：2020-05-18 13:36:55

内核源码精华帖内容汇总

展开

订阅须知

• 所有用户可根据关注领域订阅专区或所有专区

• 付费订阅：虚拟交易，一经交易不退款；若特殊情况，可3日内客服咨询

• 专区发布评论属默认订阅所评论专区（除付费小栈外）

技术专家

查看更多

飘絮絮絮丶
专家