数据报的接收过程详解---从网卡到L3层(非NAPI,即接收数据采用中断方式)

数据报的接收过程详解---从网卡到L3层(非NAPI,即接收数据采用中断方式)

来源: ChinaUnix博客 　日期： 2008.12.17 13:07　(共有0条评论)

刚来实验室的时候主要看的就是数据报在协议栈的具体传输过程,当时有过记录,但是很凌乱,最近又回头看了看相关知识和内核源代码,算是理清了思路,特整理在此.本篇笔记写的是2.4中数据报的接收过程,从网卡到网络层的具体路线,2.4中大部分网卡采用的是中断的方式接收数据(好像是从2.5以后开始支持NAPI的,不太确定),本篇笔记总结的是非NAPI,即采用中断接受数据的路线.ok,开始进入主题.

当网卡接收到一个数据报之后,产生一个中断通知内核,然后内核会调用相关的中断处理函数.一般,中断处理程序做如下工作:
1,把数据报拷贝到一个sk_buff中.
2,初始化sk_buff中的一些成员变量,为以后传输到上层用,特别是skb->protocol,标示了上层的具体协议,后面会调用相应的接收函数.
3,更新网卡的状态.
4,调用netif_rx将数据报送往上层(采用NAPI时,此处将调用netif_rx_schdule).

看netif_rx源代码之前首先要了解一下softnet_data结构.此结构是基于cpu的而不是device,即每个cpu对应一个softnet_data.
struct softnet_data
{            
        /*throttle用于拥塞控制,当拥塞时被设置,此后来的数据包都被丢弃*/
        int throttle;
        /*netif_rx返回的拥塞级别*/
        int cng_level;
        int avg_blog;
        /*input_pkt_queue是skb的队列,接收到的skb全都进入到此队列等待后续处理*/
        struct sk_buff_head input_pkt_queue;
        /*poll_list是一个双向链表,链表的成员是有接收数据等待处理的device*/
        struct list_head poll_list;
        /*net_device链表,成员为有数据报要发送的device*/
        struct net_device *output_queue;
        /*完成发送的数据包等待释放的队列*/
        struct sk_buff *completion_queue;
        /*注意,backlog_dev不是一个指针,而是一个net_device实体,代表了调用net_rx_action时的device*/
        struct net_device backlog_dev;
};

ok,了解了softnet_data结构体后接下来看netif_rx的源代码.

/*
*主要工作:
*1,初始化sk_buff的一些域值,比如数据报接收的时间截.
*2,把接收到的数据报入input_pkt_queue接收队列,并且通知内核然后触发相应的软中断,即NET_RX_SOFTIRQ.
* 2.1:当队列为空时,调用netif_rx_schedule,触发软中断.
* 2.2:当队列非空时,直接将数据报入队列,因为此时已经调用了软中断,所以无需再调用.
*3,更新相关状态信息.
*执行完后,流程来到net_rx_action
*/
int netif_rx(struct sk_buff *skb)
{
    int this_cpu = smp_processor_id();
    struct softnet_data *queue;
    unsigned long flags;
    //如果接收到的数据包时间截未设置,设置时间截
    if (skb->stamp.tv_sec == 0)
        do_gettimeofday(&skb->stamp);
    queue = &softnet_data[this_cpu];
    local_irq_save(flags); //disable irqs on local cpu
    netdev_rx_stat[this_cpu].total++;
    if (queue->input_pkt_queue.qlen = netdev_max_backlog) {
        if (queue->input_pkt_queue.qlen) {
            if (queue->throttle)
                goto drop;
enqueue:
            dev_hold(skb->dev); //即automic_inc(&(dev)->refcnt)累加设备引入计数器
            __skb_queue_tail(&queue->input_pkt_queue,skb); //把skb入input_pkt_queue队列,此处只是指针的指向,而不是数据报的拷贝,目的是节省时间.
            local_irq_restore(flags); //eable irqs on local cpu
#ifndef OFFLINE_SAMPLE
            get_sample_stats(this_cpu);
#endif
            return queue->cng_level;//返回拥塞等级
        }
     //驱动程序不断的调用netif_rx,将数据包入队操作,当qlen==0时,执行下面代码
     //如果设置了拥塞位,将其设为0
        if (queue->throttle) {
            queue->throttle = 0;
#ifdef CONFIG_NET_HW_FLOWCONTROL
            if (atomic_dec_and_test(&netdev_dropping))
                netdev_wakeup();
#endif
        }
        /*
         netif_rx_schedule主要完成两件事
         1):将接收skb的device加入"处理数据包的设备"的链表当中
         2):触发软中断函数,进行数据包接收处理,接收软中断的处理函数为net_rx_action
        */
        netif_rx_schedule(&queue->backlog_dev); //只有当input_pkt_queue为空时才调用,非空时只是将数据报入队列,因为如果队列非空,则已经调用了NET_RX_SOFTIRQ软中断,所以没必要在执行netif_rx_schedule去调用软中断了.
        goto enqueue;
    }
    /*如果队列无空闲空间,设置拥塞位*/
    if (queue->throttle == 0) {
        queue->throttle = 1;
        netdev_rx_stat[this_cpu].throttled++;
#ifdef CONFIG_NET_HW_FLOWCONTROL
        atomic_inc(&netdev_dropping);
#endif
    }
drop:
    netdev_rx_stat[this_cpu].dropped++;
    local_irq_restore(flags);
    kfree_skb(skb);
    return NET_RX_DROP;
}

看完netif_rx后有一个问题,当队列为空的时候会调用netif_rx_schedule,此函数将会把接收skb的device连接到poll_list链表,但如果队列非空时,为什么直接把数据放到input_pkt_queue里了?
想了想,应该是因为这样:因为采用NAPI技术的网卡接收到数据报后不会调用netif_rx,而直接调用netif_rx_schedule,所以调用netif_rx的都是非NAPI的网卡,那么默认的包处理函数都是process_backlog,也就是说,所有数据报的处理函数是一样的,所以当队列非空时,直接将接收到的skb放入接收队列即可.
以上纯粹是个人的理解,不知道对不对,如果不对,有知道的朋友希望告诉下,谢谢.

netif_rx返回后,由netif_rx_schedule调用软中断处理函数,所以控制权转交到net_rx_action.

/*
*接收软中断NET_RX_SOFTIRQ的处理函数
*当调用poll_list中的device时,如果网卡采用的是NAPI技术,则调用网卡自定义的poll函数,如果是非NAPI(目前大多数网卡都是采用此技术),则调用默认的process_backlog函数.
*由于此笔记研究的是非NAPI,所以控制权转交到process_backlog函数.
*/
static void net_rx_action(struct softirq_action *h)
{
    int this_cpu = smp_processor_id();
    struct softnet_data *queue = &softnet_data[this_cpu]; //获得与cpu相关的softnet_data,因为每个cpu只有一个softnet_data
    unsigned long start_time = jiffies;
    int budget = netdev_max_backlog; //默认值为300,系统每次从队列中最多取出300个skb处理
    br_read_lock(BR_NETPROTO_LOCK);
    local_irq_disable();
    while (!list_empty(&queue->poll_list)) {
        struct net_device *dev;
        //当处理时间持续超过一个时钟滴答时,会再出发一个中断NET_RX_SOFTIRQ
        if (budget = 0 || jiffies - start_time > 1)
            goto softnet_break;
        local_irq_enable();
        //取得poll_list链表中的设备
        dev = list_entry(queue->poll_list.next, struct net_device, poll_list);
        //调用设备的poll函数,处理接收数据包,采用轮寻技术的网卡,将调用它真实的poll函数
        //而对于采用传统中断处理的设备,它们调用的都将是backlog_dev的process_backlog函数
        //如果一次poll未处理完全部数据报,则将device移至链表尾部,等待下一次调用.
        if (dev->quota = 0 || dev->poll(dev, &budget)) {
            //由于要对softnet_data进行操作,则必须禁止中断.
            local_irq_disable();
            //把device从表头移除,移到链表尾
            list_del(&dev->poll_list);
            list_add_tail(&dev->poll_list, &queue->poll_list);
            if (dev->quota  0)
                dev->quota += dev->weight;
            else
                dev->quota = dev->weight;
        } else {
            dev_put(dev); //当device中的数据报被处理完毕(网卡采用NAPI技术时),递减device的引用计数 dcreases the reference count .
            local_irq_disable();
        }
    }
    local_irq_enable();
    br_read_unlock(BR_NETPROTO_LOCK);
    return;
softnet_break:
    netdev_rx_stat[this_cpu].time_squeeze++;
    //如果此时又有中断发生,出发NET_RX_SOFTIRQ中断,再此调用net_rx_action函数.
    __cpu_raise_softirq(this_cpu, NET_RX_SOFTIRQ);
    local_irq_enable();
    br_read_unlock(BR_NETPROTO_LOCK);
}

软中断处理函数的主体既是调用数据报的处理函数,网卡采用NAPI技术的情况下,则调用device本身定义的poll函数,如果是非NAPI,则调用的是默认的process_backlog函数,既然本笔记研究的是非NAPI的情况,那么下面来研究下process_backlog函数.

/*
*把数据报从input_pkt_queue中去出来后,交由netif_receive_skb处理,netif_receive_skb为真正的包处理函数.
*注意,无论是基于NAPI还是非NAPI,最后的包处理函数都是netif_receive_skb.
*/
static int process_backlog(struct net_device *blog_dev, int *budget)
{
    int work = 0;
   
    //quota为一次处理数据包的数量,blog_dev->quota的值由netif_rx_schedule初始化为全局变量weight_p的值,默认值为64
    int quota = min(blog_dev->quota, *budget);
    int this_cpu = smp_processor_id();
    struct softnet_data *queue = &softnet_data[this_cpu];
    unsigned long start_time = jiffies;
  //循环取出skb,交由netif_receive_skb处理,直至队列为空
    for (;;) {
        struct sk_buff *skb;
        struct net_device *dev;
        local_irq_disable();
        skb = __skb_dequeue(&queue->input_pkt_queue);
        if (skb == NULL)
            goto job_done;
        local_irq_enable();
        dev = skb->dev;
        //注意此处,取出数据报后,直接交由netif_receive_skb处理.
        netif_receive_skb(skb);
        dev_put(dev); //dev引用计数-1
        work++;
        if (work >= quota || jiffies - start_time > 1)
        break;
#ifdef CONFIG_NET_HW_FLOWCONTROL
        if (queue->throttle && queue->input_pkt_queue.qlen  no_cong_thresh ) {
            if (atomic_dec_and_test(&netdev_dropping)) {
                queue->throttle = 0;
                netdev_wakeup();
                break;
            }
        }
#endif
    }
    //更新quota
    blog_dev->quota -= work;
    *budget -= work;
    return -1;
job_done:
    blog_dev->quota -= work;
    *budget -= work;
    list_del(&blog_dev->poll_list);
    clear_bit(__LINK_STATE_RX_SCHED, &blog_dev->state);
    if (queue->throttle) {
        queue->throttle = 0;
#ifdef CONFIG_NET_HW_FLOWCONTROL
        if (atomic_dec_and_test(&netdev_dropping))
            netdev_wakeup();
#endif
    }
    local_irq_enable();
    return 0;
}
注意,就像注释中所说,无论NAPI还是非NAPI,最后的包处理函数都是netif_receive_skb.所以此函数比较重要.下面来分析下此函数.
/*
  netif_receive_skb作用:
    对每一个接收到的skb,到已注册的协议类型中去匹配,先是匹配ptype_all链表,ptype_all中注册的struct packet_type表示要接收处理所有协议的数据,对于
    匹配到的struct packet_tpye结构(dev为NULL或者dev等于skb的dev),调用其func成员,把skb传递给它处理.
    匹配完ptype_all后,再匹配ptype_base数组中注册的协议类型,skb有一个成员protocol,其值即为以太网首部中的帧类型,在ptype_base中匹配到协议相同,并且
    dev符合要求的,调用其func成员即可.
    此函数中数据报的流向:sniffer(如果有)->Diverter(分流器)->bridge(如果有)->l3协议的处理函数(比如ip协议的ip_rcv)
*/
int netif_receive_skb(struct sk_buff *skb)
{
    //packet_type结构体见下面
    struct packet_type *ptype, *pt_prev;
    int ret = NET_RX_DROP;
    unsigned short type = skb->protocol;
   
    //如果数据包没设置时间截,设置之
    if (skb->stamp.tv_sec == 0)
        do_gettimeofday(&skb->stamp);
    skb_bond(skb); //使skb->dev指向主设备,多个interfaces可以在一起集中管理,这时候要有个头头管理这些接口,如果skb对应的device来自这样一个group,
     //则传递到L3层之前应使skb->dev指向master
    netdev_rx_stat[smp_processor_id()].total++;
#ifdef CONFIG_NET_FASTROUTE
    if (skb->pkt_type == PACKET_FASTROUTE) {
        netdev_rx_stat[smp_processor_id()].fastroute_deferred_out++;
        return dev_queue_xmit(skb);
    }
#endif
    skb->h.raw = skb->nh.raw = skb->data;
    pt_prev = NULL;
   
    //ptype_all是双向链表,ptpye_base是一个哈希表
    //初始化时协议类型为ETH_P_ALL时,将packet_type结构加入到ptype_all列表中   
    //如果不是，模15后加到ptype_base数组中，此数组相当于hash链表
   
    //这里针对协议类型为ETH_P_ALL的情况进行处理,对于ip协议来说
    //类型定义为ETH_P_IP,因此不在这里处理
   
    for (ptype = ptype_all; ptype; ptype = ptype->next) {
  //处理器处理所有的网络设备接收的包或找到设备匹配的包处理器
      if (!ptype->dev || ptype->dev == skb->dev) {
            if (pt_prev) {
                //老版本内核时
                if (!pt_prev->data) {
                 /* Deliver skb to an old protocol, which is not threaded well
                       or which do not understand shared skbs.
                */
                    ret = deliver_to_old_ones(pt_prev, skb, 0);
                } else {
                    //发给相应的处理函数,下面两条相当于deliver_skb
                    //有协议对skb处理,所以use加1
                    atomic_inc(&skb->users);
                    //传递的是skb的指针,所以可以看出是原数据报本身.即如果在此修改skb,则会影响到后面的数据流向.
                    ret = pt_prev->func(skb, skb->dev, pt_prev);
                }
            }
            pt_prev = ptype;
        }
    }
#ifdef CONFIG_NET_DIVERT
  /*如果配置有DIVERT(分流器),则交由分流器处理*/
    if (skb->dev->divert && skb->dev->divert->divert)
        ret = handle_diverter(skb);
#endif /* CONFIG_NET_DIVERT */
    /*如果配置有BRIDGE或者有BRIDGE模块,则交由桥处理*/        
#if defined(CONFIG_BRIDGE) || defined(CONFIG_BRIDGE_MODULE)
    if (skb->dev->br_port != NULL &&
     br_handle_frame_hook != NULL) {
        return handle_bridge(skb, pt_prev);
    }
#endif
  //这里针对各种协议进行处理,eg:ip包的类型为ETH_P_IP,因此在这里处理
    //&15的意思是模15
    for (ptype=ptype_base[ntohs(type)&15];ptype;ptype=ptype->next) {
        if (ptype->type == type &&
         (!ptype->dev || ptype->dev == skb->dev)) {
            if (pt_prev) { //pt_prev指向具体的协议类型
                if (!pt_prev->data) {
                    ret = deliver_to_old_ones(pt_prev, skb, 0);
                } else {
                    atomic_inc(&skb->users);
                    ret = pt_prev->func(skb, skb->dev, pt_prev);
                }
            }
            pt_prev = ptype;
        }
    }
  //1:当上面的两个数组只有一个元素时
  //2:访问上面两个数组的最后一个ptype_type,执行下面的语句
    if (pt_prev) {
        if (!pt_prev->data) {
            ret = deliver_to_old_ones(pt_prev, skb, 1);
        } else {
            //当只有一个协议的时候,user不加1
            ret = pt_prev->func(skb, skb->dev, pt_prev);
        }
    } else { ///////表示搜索完ptype_all和ptype_base后没找到匹配的,free掉skb
        kfree_skb(skb);
        /* Jamal, now you will not able to escape explaining
         * me how you were going to use this. :-)
         */
        ret = NET_RX_DROP;
    }
  return ret;
}
其中packet_type的定义如下:
struct packet_type
    {
        unsigned short type;
        //一般的dev设置为NULL,表示将接收从任何设备接收的数据包
        struct net_device *dev;
        int (*func)(struct sk_buff*,strcut net_device*,
                                                                struct packet_type *); //接收处理函数
        void *data;//private to the packet type
        struct packet_type *next;
    };
注意,在网络初始化代码里有这么一条语句:struct packet_type * ptype_all = NULL.就是说ptype_all在初始化的时候为空,用于指向Eth_P_ALL类型的packet_type结构,注册在这里的函数,可以接收到所有的数据报.包括输出的数据包,看代码可知,由于传递的是一个指针,所以在此修改skb的一切操作将会影响后面的处理过程.
如果想要在数据包传递到网络层之前对数据报进行处理,则可以考虑的地点可以有如下几个:
sniffer,diverter,bridge. 其中sniffer本人已经亲自实现过,即在ptype_all中注册自己的函数,可以接收到所有出去或者进入的数据包.
最后,数据包会传到l3层,如果是ip协议,则相应的处理函数为ip_rcv,到此数据报从网卡到l3层的接收过程已经完毕.即总的路线是:netif_rx-->net_rx_action-->process_backlog-->netif_receive_skb-->sniffer(如果有)-->diverter(如果有)-->bridge(如果有)-->ip_rcv(或者其他的l3层协议处理函数)
ps1:弄了好几天,终于算把这个数据包的接收过程系统的过了一遍,看过源代码,感觉收获不小,但是觉得知道的还不是很透彻,其中的设计理念等以后有机会一定要好好的研究研究.由于本次涉及到的东西相对来说比较多,而且涉及到内核网络部分,本人很菜,难免有理解错误的地方,如果有错误,请看到的朋友能够及时给予指正,不胜感激.
ps2:从在网上查找资料,看相关书籍到通读这一过程的源代码,本人花费了很多时间,因此对本篇笔记很是喜爱和重视,如果有需要转载的朋友,请注明来自hmily8023.cublog.cn,谢谢.