首页 > 自动化运维 > > PF_RING实现分析(2)
4、mmap操作
用户态的接下来调用:
进行内存映射。
同样地,内核调用相应的ring_mmap进行处理。
Ring选项结构通过ring_sk宏与sk 建立关联
pfr->ring_memory 即为分配的环形队列空间。所以,要mmap操作,实际上就是调用remap_pfn_range函数把pfr->ring_memory 映射到用户空间即可。这个函数的原型为:
关于remap_pfn_range函数的进一步说明,可以参考LDD3,上面有详细说明和现成的例子。
实际上的内存映射工作,是由do_memory_mmap来完成的,这个函数实际上基本就是remap_pfn_range的包裹函数。
不过因为系统支持dna等技术,相应的mode参数有些变化,这里只分析了最基本的方法:mode == 0
嗯,跳过了太多的细节,不过其mmap最核心的东东已经呈现出来。
如果要共享内核与用户空间内存,这倒是个现成的可借鉴的例子。
5、数据包的入队操作
做到这一步,准备工作基本上就完成了。因为PF_RING在初始化中,注册了prot_hook。其func指针指向packet_rcv函数:
当数据报文进入Linux网络协议栈队列时,netif_receive_skb会遍历这些注册的Hook:
相应的Hook函数得到调用:
packet_rcv随之执行环形队列的入队操作:
static int skb_ring_handler(struct sk_buff *skb, //要捕获的数据包
u_char recv_packet, //数据流方向,>0表示是进入(接收)方向
u_char real_skb /* 1=real skb, 0=faked skb */ ,
short channel_id) //通道ID
{
struct sock *skElement;
int rc = 0, is_ip_pkt;
struct list_head *ptr;
struct pfring_pkthdr hdr;
int displ;
struct sk_buff *skk = NULL;
struct sk_buff *orig_skb = skb;
#ifdef PROFILING
uint64_t rdt = _rdtsc(), rdt1, rdt2;
#endif
//skb合法检查,包括数据流的方向
if((!skb) /* Invalid skb */
||((!enable_tx_capture) && (!recv_packet))) {
/*
An outgoing packet is about to be sent out
but we decided not to handle transmitted
packets.
*/
return(0);
}
#if defined(RING_DEBUG)
if(1) {
struct timeval tv;
skb_get_timestamp(skb, &tv);
printk
("[PF_RING] skb_ring_handler() [skb=%p][%u.%u][len=%d][dev=%s][csum=%u]\n",
skb, (unsigned int)tv.tv_sec, (unsigned int)tv.tv_usec,
skb->len,
skb->dev->name == NULL ? "<NULL>" : skb->dev->name,
skb->csum);
}
#endif
//如果通道ID未指定,根据进入的报文设备索引,设定之
#if(LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,21))
if(channel_id == RING_ANY_CHANNEL /* Unknown channel */ )
channel_id = skb->iif; /* Might have been set by the driver */
#endif
#if defined (RING_DEBUG)
/* printk("[PF_RING] channel_id=%d\n", channel_id); */
#endif
#ifdef PROFILING
rdt1 = _rdtsc();
#endif
if(recv_packet) {
/* Hack for identifying a packet received by the e1000 */
if(real_skb)
displ = SKB_DISPLACEMENT;
else
displ = 0; /* Received by the e1000 wrapper */
} else
displ = 0;
//解析数据报文,并判断是否为IP报文
is_ip_pkt = parse_pkt(skb, displ, &hdr);
//分片处理,是一个可选的功能项,事实上,对大多数包捕获工具而言,它们好像都不使用底层库来完成这一功能
/* (de)Fragmentation <fusco@ntop.org> */
if(enable_ip_defrag
&& real_skb && is_ip_pkt && recv_packet && (ring_table_size > 0)) {
} else {
#if defined (RING_DEBUG)
printk("[PF_RING] Do not seems to be a fragmented ip_pkt[iphdr=%p]\n",
iphdr);
#endif
}
}
}
//按惯例,在报文的捕获首部信息中记录捕获的时间戳
/* BD - API changed for time keeping */
#if(LINUX_VERSION_CODE < KERNEL_VERSION(2,6,14))
if(skb->stamp.tv_sec == 0)
do_gettimeofday(&skb->stamp);
hdr.ts.tv_sec = skb->stamp.tv_sec, hdr.ts.tv_usec = skb->stamp.tv_usec;
#elif(LINUX_VERSION_CODE < KERNEL_VERSION(2,6,22))
if(skb->tstamp.off_sec == 0)
__net_timestamp(skb);
hdr.ts.tv_sec = skb->tstamp.off_sec, hdr.ts.tv_usec =
skb->tstamp.off_usec;
#else /* 2.6.22 and above */
if(skb->tstamp.tv64 == 0)
__net_timestamp(skb);
hdr.ts = ktime_to_timeval(skb->tstamp);
#endif
//除了时间,还有长度,熟悉libpcap的话,这些操作应该很眼熟
hdr.len = hdr.caplen = skb->len + displ;
/* Avoid the ring to be manipulated while playing with it */
read_lock_bh(&ring_mgmt_lock);
/* 前面在创建sk时,已经看过ring_insert的入队操作了,现在要检查它的成员
* 它们的关系是,通过ring_table的成员,获取到element,它里面封装了sk,
*通过ring_sk宏,就可以得到ring_opt指针
*/
list_for_each(ptr, &ring_table) {
struct ring_opt *pfr;
struct ring_element *entry;
entry = list_entry(ptr, struct ring_element, list);
skElement = entry->sk;
pfr = ring_sk(skElement);
//看来要加入社团,条件还是满多的,pfr不能为空,未指定集群cluster_id,槽位不能为空,方向要正确,绑定的网络设备
//得对上号
//另一种可能就是对bonding的支持,如果设备是从属设备,则应校验其主设备
if((pfr != NULL)
&& (pfr->cluster_id == 0 /* No cluster */ )
&& (pfr->ring_slots != NULL)
&& is_valid_skb_direction(pfr->direction, recv_packet)
&& ((pfr->ring_netdev == skb->dev)
|| ((skb->dev->flags & IFF_SLAVE)
&& (pfr->ring_netdev == skb->dev->master)))) {
/* We've found the ring where the packet can be stored */
/* 从新计算捕获帧长度,是因为可能因为巨型帧的出现——超过了桶能容纳的长度 */
int old_caplen = hdr.caplen; /* Keep old lenght */
hdr.caplen = min(hdr.caplen, pfr->bucket_len);
/* 入队操作 */
add_skb_to_ring(skb, pfr, &hdr, is_ip_pkt, displ, channel_id);
hdr.caplen = old_caplen;
rc = 1; /* Ring found: we've done our job */
}
}
/* [2] Check socket clusters */
list_for_each(ptr, &ring_cluster_list) {
ring_cluster_element *cluster_ptr;
struct ring_opt *pfr;
cluster_ptr = list_entry(ptr, ring_cluster_element, list);
if(cluster_ptr->cluster.num_cluster_elements > 0) {
u_int skb_hash = hash_pkt_cluster(cluster_ptr, &hdr);
skElement = cluster_ptr->cluster.sk[skb_hash];
if(skElement != NULL) {
pfr = ring_sk(skElement);
if((pfr != NULL)
&& (pfr->ring_slots != NULL)
&& ((pfr->ring_netdev == skb->dev)
|| ((skb->dev->flags & IFF_SLAVE)
&& (pfr->ring_netdev ==
skb->dev->master)))
&& is_valid_skb_direction(pfr->direction, recv_packet)
) {
/* We've found the ring where the packet can be stored */
add_skb_to_ring(skb, pfr, &hdr,
is_ip_pkt, displ,
channel_id);
rc = 1; /* Ring found: we've done our job */
}
}
}
}
read_unlock_bh(&ring_mgmt_lock);
#ifdef PROFILING
rdt1 = _rdtsc() - rdt1;
#endif
#ifdef PROFILING
rdt2 = _rdtsc();
#endif
/* Fragment handling */
if(skk != NULL)
kfree_skb(skk);
if(rc == 1) {
if(transparent_mode != driver2pf_ring_non_transparent) {
rc = 0;
} else {
if(recv_packet && real_skb) {
#if defined(RING_DEBUG)
printk("[PF_RING] kfree_skb()\n");
#endif
kfree_skb(orig_skb);
}
}
}
#ifdef PROFILING
rdt2 = _rdtsc() - rdt2;
rdt = _rdtsc() - rdt;
#if defined(RING_DEBUG)
printk
("[PF_RING] # cycles: %d [lock costed %d %d%%][free costed %d %d%%]\n",
(int)rdt, rdt - rdt1,
(int)((float)((rdt - rdt1) * 100) / (float)rdt), rdt2,
(int)((float)(rdt2 * 100) / (float)rdt));
#endif
#endif
//printk("[PF_RING] Returned %d\n", rc);
return(rc); /* 0 = packet not handled */
}
上面跳过了对cluster(集群)的分析,PF_RING允许同时对多个接口捕获报文,而并不是一个。这就是集群。看一下它用户态的注释就一目了然了:
进一步的入队操作,是通过add_skb_to_ring来完成的:
槽位的计算:
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/69946831/viewspace-3136325/ ,如需转载,请注明出处,否则将追究法律责任。