PF_RING实现分析（1）

2020-05-22 00:00:00 用户 队列 内核 分配 初始化

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

独孤九贱

内核版本：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 结构：

  1. /*
  2. * ring_insert()
  3. *
  4. * store the sk in a new element and add it
  5. * to the head of the list.
  6. */
  7. static inline void ring_insert(struct sock *sk)
  8. {
  9.   struct ring_element *next;
  10.   struct ring_opt *pfr;

  12. #if defined(RING_DEBUG)
  13.   printk("[PF_RING] ring_insert()\n");
  14. #endif

  16. //分配element
  17.   next = kmalloc(sizeof(struct ring_element), GFP_ATOMIC);
  18.   if(next != NULL) {
  19.     //记录sk
  20.     next->sk = sk;
  21.     write_lock_bh(&ring_mgmt_lock);
  22.     //入队
  23.     list_add(&next->list, &ring_table);
  24.     write_unlock_bh(&ring_mgmt_lock);
  25.   } else {
  26.     if(net_ratelimit())
  27.       printk("[PF_RING] net_ratelimit() failure\n");
  28.   }

  30.   //累计使用计数器
  31.   ring_table_size++;
  32.   //ring_proc_add(ring_sk(sk));
  33.   //记录进程PID
  34.   pfr = (struct ring_opt *)ring_sk(sk);
  35.   pfr->ring_pid = current->pid;
  36. }

相关文章