Linux网络-数据包的接收流程（基于RTL8139网卡驱动程序）

本文将介绍Linux系统中，基于RTL8139网卡驱动程序，是如何一步一步将接收到的数据包传送到内核的网络协议栈的。

下图展示了数据包（packet）如何进入内存，并被内核的网络模块开始处理：

                   +-----+|     |                            Memroy
+--------+   1     |     |  2  DMA     +--------+--------+--------+--------+
| Packet |-------->| NIC |------------>| Packet | Packet | Packet | ...... |
+--------+         |     |             +--------+--------+--------+--------+|     |<--------++-----+         ||            +---------------+|                            |3 | Raise IRQ                  | Disable IRQ|                          5 ||                            |↓                            |+-----+                   +------------+|     |  Run IRQ handler  |            || CPU |------------------>| NIC Driver ||     |       4           |            |+-----+                   +------------+|6  | Raise soft IRQ|↓

1.数据包从外部网络传送到物理网卡。如果目的地址不是该网卡，且该网卡没有开启混杂模式，该包会被网卡丢弃。
2.网卡将数据包以DMA的方式写进指定的内存地址。该地址是由网卡驱动程序分配并初始化的。
3.数据来了之后，网卡产生一个硬件中断（IRQ）告诉CPU。
4.CPU会根据中断向量表，调用中断处理函数，这个中断处理函数会调用网卡驱动程序中的中断函数。
5.驱动先禁用网卡的中断，因为驱动程序已经知道内存中有数据了，告诉网卡驱动程序下次再有数据，直接写进内存中，不用再通知CPU了，这样可以提高效率。避免CPU被不停的中断打扰。
6.启用软中断。这步结束后，硬件中断就结束返回了。由于硬中断在执行程序的过程中，不能被中断，所以如果执行时间过长，会导致CPU没法响应其他硬件的中断，于是引入了软中断，这样可以将硬中断处理函数中比较耗时部分，交给软中断处理函数里慢慢处理。

软中断会触发内核网络模块中的软中断处理函数，流程如下：

                                                     +-----+17      |     |+----------->| NIC ||            |     ||Enable IRQ  +-----+||+------------+                                      Memroy|            |        Read           +--------+--------+--------+--------++--------------->| NIC Driver |<--------------------- | Packet | Packet | Packet | ...... ||                |            |          9            +--------+--------+--------+--------+|                +------------+|                      |    |        skbPoll | 8      Raise softIRQ | 6  +-----------------+|                      |             10       ||                      ↓                      ↓+---------------+  Call  +-----------+        +------------------+        +--------------------+  12  +---------------------+| net_rx_action |<-------| ksoftirqd |        | napi_gro_receive |------->| enqueue_to_backlog |----->| CPU input_pkt_queue |+---------------+   7    +-----------+        +------------------+   11   +--------------------+      +---------------------+|                                                      | 1314 |        + - - - - - - - - - - - - - - - - - - - - - - +↓        ↓+--------------------------+    15      +------------------------+| __netif_receive_skb_core |----------->| packet taps(AF_PACKET) |+--------------------------+            +------------------------+|| 16↓+-----------------+| protocol layers |+-----------------+

7 内核中的ksoftirqd进程专门负责软中断的处理，当它收到软中断后，就会调用相应软中断所对应的处理函数，对于上面第6步中是网卡驱动模块抛出的软中断，ksoftirqd会调用网络模块的net_rx_action函数
8 net_rx_action函数会调用网卡驱动程序里的poll函数来一个一个处理数据包。
9 在poll函数中，网卡驱动程序一个接一个读取网卡写到内存中的数据包，内存中的数据包的格式只有驱动程序知道。
10 驱动程序将从内存中读取到的数据包转换成内核网络模块能识别的格式skb格式，然后调用napi_gro_receive函数
11 napi_gro_receive会处理GRO相关的内容，也就是将可以合并的数据包进行合并，这样就只需要调用一次协议栈。然后判断是否开启了RPS，如果开启了，将会调用enqueue_to_backlog.
12 在enqueue_to_backlog函数中，会将数据包放入CPU的softnet_data结构体的input_pkt_queue中，然后返回，如果input_pkt_queue满了的话，该数据包将会被丢弃，queue的大小可以通过net.core.netdev_max_backlog来配置。
13 CPU会接着在自己的软中断上下文中调用__netif_receive_skb_core函数处理自己input_pkt_queue里的网络数据。
14 如果没开启RPS，napi_gro_receive会直接调用__netif_receive_skb_core
15 看是不是有AF_PACKET类型的socket（也就是我们常说的原始套接字），如果有的话，拷贝一份数据给它。tcpdump抓包就是抓的这里的包。
16 调用协议栈相应的函数，将数据包交给协议栈处理。
17 待内存中的所有数据包被处理完成后（即poll函数执行完成），启用网卡的硬中断，这样下次网卡再收到数据的时候就会通知CPU。

enqueue_to_backlog函数也会被netif_rx函数调用，而netif_rx正是网络堆栈接收从lo设备发送过来的数据包时所要调用的函数

以上分析参考地址：
点击查看原文

以上是大致分析了一下数据包是如何从网卡传送到内核网络协议栈的。那么下面我们就分析网卡驱动程序里面的相关函数：

一.中断函数
首先是中断函数rtl8139_interrupt（），当硬件中断后，CPU会调用网卡驱动程序的该函数，rtl8139_interrupt函数是在rtl8139_open函数中注册的：

static int rtl8139_open (struct net_device *dev)
{
。。。/* 注册中断处理函数 ，中断号为共享 */retval = request_irq (dev->irq, rtl8139_interrupt, IRQF_SHARED, dev->name, dev);
。。。
}

我们先来看中断函数：


/* The interrupt handler does all of the Rx thread work and cleans upafter the Tx thread. */
static irqreturn_t rtl8139_interrupt (int irq, void *dev_instance)
{/* 参数dev_instance是在上面注册中断处理函数的时候传入的 */struct net_device *dev = (struct net_device *) dev_instance;/* tp 为网卡驱动自定义的驱动特有的数据，和dev一起分配的 */struct rtl8139_private *tp = netdev_priv(dev);void __iomem *ioaddr = tp->mmio_addr;u16 status, ackstat;int link_changed = 0; /* avoid bogus "uninit" warning */int handled = 0;/* 对驱动数据加锁*/spin_lock (&tp->lock);/*读中断状态寄存器，获取中断状态*/status = RTL_R16 (IntrStatus);/* shared irq? *//* 这时由共享此中断号的其它设备产生的中断 */if (unlikely((status & rtl8139_intr_mask) == 0))goto out;handled = 1;/* h/w no longer present (hotplug?) or major error, bail *//* 硬件错误 */if (unlikely(status == 0xFFFF))goto out;/* close possible race's with dev_close *//* 设备已关闭*/if (unlikely(!netif_running(dev))) {/* 屏蔽所有中断*/RTL_W16 (IntrMask, 0);goto out;}/* Acknowledge all of the current interrupt sources ASAP, butan first get an additional status bit from CSCR. */if (unlikely(status & RxUnderrun))link_changed = RTL_R16 (CSCR) & CSCR_LinkChangeBit;ackstat = status & ~(RxAckBits | TxErr);if (ackstat)RTL_W16 (IntrStatus, ackstat);/* Receive packets are processed by poll routine.If not running start it now. *//* 下一步处理数据包到达事件 */if (status & RxAckBits){if (napi_schedule_prep(&tp->napi)) {RTL_W16_F (IntrMask, rtl8139_norx_intr_mask);/* 这个函数的分析在下面 */__napi_schedule(&tp->napi);    //这里采用了NAPI新机制，暂时不详细说明这个机制，会新开一篇文章，详细讲解一下}}/* 以下都是一些检查，在此不做分析 *//* Check uncommon events with one test. */if (unlikely(status & (PCIErr | PCSTimeout | RxUnderrun | RxErr)))rtl8139_weird_interrupt (dev, tp, ioaddr,status, link_changed);if (status & (TxOK | TxErr)) {/* 发送完成事件处理 ，下面会分析*/rtl8139_tx_interrupt (dev, tp, ioaddr);if (status & TxErr)RTL_W16 (IntrStatus, TxErr);}out:spin_unlock (&tp->lock);netdev_dbg(dev, "exiting interrupt, intr_status=%#4.4x\n",RTL_R16(IntrStatus));return IRQ_RETVAL(handled);
}

__napi_schedule（）函数的分析：

void __napi_schedule(struct napi_struct *n)
{unsigned long flags;local_irq_save(flags);  //这里应该是保存中断标志位____napi_schedule(&__get_cpu_var(softnet_data), n);  //去这个函数看看local_irq_restore(flags);   //回复中断标志位
}

____napi_schedule（）看看这个函数：

static inline void ____napi_schedule(struct softnet_data *sd,struct napi_struct *napi)
{list_add_tail(&napi->poll_list, &sd->poll_list);  //将当前设备加入CPU相关全局队列softnet_data的轮询设备列表中/* 调用函数产生网络接收软中断。 */__raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

二. 发送完成事件处理
下面我们分析rtl8139_tx_interrupt函数：

static void rtl8139_tx_interrupt (struct net_device *dev,struct rtl8139_private *tp,void __iomem *ioaddr)
{unsigned long dirty_tx, tx_left;assert (dev != NULL);assert (ioaddr != NULL);/*dirty_tx是最近发送数据包时，没有经中断处理的最早数据包所对应的发送描述符*/dirty_tx = tp->dirty_tx;/* cur_tx是最近发送完成的最后一个数据包对应的发送描述符，所以在此次中断中要处理的就是和dirty_tx之间的发送描述符*/tx_left = tp->cur_tx - dirty_tx;while (tx_left > 0) {/* 环形缓冲区，最大为NUM_TX_DESC，取模得到真实值*/int entry = dirty_tx % NUM_TX_DESC;int txstatus;/*当前发送描述符的发送状态(一个寄存器为32bit)*/ txstatus = RTL_R32 (TxStatus0 + (entry * sizeof (u32)));/*还没有发送*/if (!(txstatus & (TxStatOK | TxUnderrun | TxAborted)))break;  /* It still hasn't been Txed *//* Note: TxCarrierLost is always asserted at 100mbps. */if (txstatus & (TxOutOfWindow | TxAborted)) {/* There was an major error, log it. */netif_dbg(tp, tx_err, dev, "Transmit error, Tx status %08x\n",txstatus);dev->stats.tx_errors++;if (txstatus & TxAborted) {dev->stats.tx_aborted_errors++;RTL_W32 (TxConfig, TxClearAbt);RTL_W16 (IntrStatus, TxErr);wmb();}if (txstatus & TxCarrierLost)dev->stats.tx_carrier_errors++;if (txstatus & TxOutOfWindow)dev->stats.tx_window_errors++;} else {if (txstatus & TxUnderrun) {/* Add 64 to the Tx FIFO threshold. */if (tp->tx_flag < 0x00300000)tp->tx_flag += 0x00020000;dev->stats.tx_fifo_errors++;}dev->stats.collisions += (txstatus >> 24) & 15;dev->stats.tx_bytes += txstatus & 0x7ff;dev->stats.tx_packets++;}dirty_tx++;tx_left--;}#ifndef RTL8139_NDEBUGif (tp->cur_tx - dirty_tx > NUM_TX_DESC) {netdev_err(dev, "Out-of-sync dirty pointer, %ld vs. %ld\n",dirty_tx, tp->cur_tx);dirty_tx += NUM_TX_DESC;}
#endif /* RTL8139_NDEBUG *//* only wake the queue if we did work, and the queue is stopped */if (tp->dirty_tx != dirty_tx) {tp->dirty_tx = dirty_tx;mb();netif_wake_queue (dev);}
}

三.软中断处理函数

由于在前面的中断处理程序中调用了__raise_softirq_irqoff(NET_RX_SOFTIRQ)，CPU会在中断处理完成后的适当的时候调用软中断处理函数，也就是我们在系统初始化的过程中注册的net_rx_action函数。

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();/* 处理poll_list上关联的每一个设备*/while (!list_empty(&sd->poll_list)) {struct napi_struct *n;int work, weight;/* If softirq window is exhuasted then punt.* Allow this to run for 2 jiffies since which will allow* an average latency of 1.5/HZ.*/if (unlikely(budget <= 0 || time_after(jiffies, time_limit))) //如果收到数据的总数到了300个goto softnet_break;local_irq_enable();/* Even though interrupts have been re-enabled, this* access is safe because interrupts can only add new* entries to the tail of this list, and only ->poll()* calls can remove this head entry from the list.*/n = list_first_entry(&sd->poll_list, struct napi_struct, poll_list);have = netpoll_poll_lock(n);weight = n->weight;/* This NAPI_STATE_SCHED test is for avoiding a race* with netpoll's poll_napi().  Only the entity which* obtains the lock and sees NAPI_STATE_SCHED set will* actually make the ->poll() call.  Therefore we avoid* accidentally calling ->poll() when NAPI is not scheduled.*/work = 0;/* 调用每个设备的pool方法接收数据*/if (test_bit(NAPI_STATE_SCHED, &n->state)) {work = n->poll(n, weight);trace_napi_poll(n);}WARN_ON_ONCE(work > weight);budget -= work; //更新budget，这样能控制总收到的数据local_irq_disable();/* Drivers must not modify the NAPI state if they* consume the entire weight.  In such cases this code* still "owns" the NAPI instance and therefore can* move the instance around on the list at-will.*/if (unlikely(work == weight)) {/* 设备运行出错，或自己退出poll_list，就删除它*/if (unlikely(napi_disable_pending(n))) {local_irq_enable();napi_complete(n);local_irq_disable();} else/* 该设备还有要接收的数据没被处理,因为轮询算法被移动到poll_llst尾部等待处理*/list_move_tail(&n->poll_list, &sd->poll_list);}netpoll_poll_unlock(have);}
out:net_rps_action_and_irq_enable(sd);#ifdef CONFIG_NET_DMA/** There may not be any more sk_buffs coming right now, so push* any pending DMA copies to hardware*/dma_issue_pending_all();
#endifreturn;softnet_break:sd->time_squeeze++;__raise_softirq_irqoff(NET_RX_SOFTIRQ);goto out;
}

通常，在网卡收发数据的时候，需要维护一个缓冲区队列，来缓存可能存在的突发数据，类似于前面的DMA环形缓冲区。队列层中，包含了一个叫做struct softnet_data：

/** Incoming packets are placed on per-cpu queues*/
struct softnet_data {struct Qdisc        *output_queue;struct Qdisc        **output_queue_tailp;struct list_head    poll_list;struct sk_buff      *completion_queue;struct sk_buff_head process_queue;/* stats */unsigned int        processed;unsigned int        time_squeeze;unsigned int        cpu_collision;unsigned int        received_rps;#ifdef CONFIG_RPSstruct softnet_data *rps_ipi_list;/* Elements below can be accessed between CPUs for RPS */struct call_single_data csd ____cacheline_aligned_in_smp;struct softnet_data *rps_ipi_next;unsigned int        cpu;unsigned int        input_queue_head;unsigned int        input_queue_tail;
#endifunsigned        dropped;struct sk_buff_head input_pkt_queue;struct napi_struct  backlog;
};

下一步进入设备的poll函数。需要注意的是，如果是NAPI的网卡驱动的话，poll函数是在驱动中注册的，驱动实现的；如果是非 NAPI的话，就是内核定义的process_backlog函数，至于process_backlog是如何添加到poll_list中的，这里暂时不管，先看看8139驱动的poll 函数是如何实现的。

四.8139 poll函数实现

static int rtl8139_poll(struct napi_struct *napi, int budget)
{struct rtl8139_private *tp = container_of(napi, struct rtl8139_private, napi);struct net_device *dev = tp->dev;void __iomem *ioaddr = tp->mmio_addr;int work_done;spin_lock(&tp->rx_lock);work_done = 0;/* 在 rtl8139_rx中将接送到的数据拷贝出来并传递给上层协议驱动。*/if (likely(RTL_R16(IntrStatus) & RxAckBits))work_done += rtl8139_rx(dev, tp, budget);/*说明没有多余的数据到达，则恢复接收中断，并把此设备从poll_list中清除*/if (work_done < budget) {unsigned long flags;/** Order is important since data can get interrupted* again when we think we are done.先关中断，在写中断屏蔽位*/spin_lock_irqsave(&tp->lock, flags);__napi_complete(napi);RTL_W16_F(IntrMask, rtl8139_intr_mask);spin_unlock_irqrestore(&tp->lock, flags);}spin_unlock(&tp->rx_lock);return work_done;
}

从rtl8139_rx的代码也可以看出，当数据包接收出错或者是没有更多的数据包可以接收时，work_done才不会达到budget，这时，应该让网卡重新回到中断的状态，以等待数据包的到来。另外一种情况就是work_done等于budget，很可能是因为还有数据包要接收，所以在net_rx_action函数中，只是把该网卡设备移到队列的尾部，以期待在下次循环中再次调用其poll函数。

下面看rtl8139_rx的实现
五.rtl8139_rx的实现

static int rtl8139_rx(struct net_device *dev, struct rtl8139_private *tp,int budget)
{void __iomem *ioaddr = tp->mmio_addr;int received = 0;/* 网卡不断的把数据放进环形接收缓冲区， CPU读出来的时候，读到哪里的顺序需要自己维护，tp->cur_rx记录上次读到哪里，这里将接着从上次的地方拷贝。*/unsigned char *rx_ring = tp->rx_ring;unsigned int cur_rx = tp->cur_rx;unsigned int rx_size = 0;netdev_dbg(dev, "In %s(), current %04x BufAddr %04x, free to %04x, Cmd %02x\n",__func__, (u16)cur_rx,RTL_R16(RxBufAddr), RTL_R16(RxBufPtr), RTL_R8(ChipCmd));/*轮询寄存器，当ChipCmd RxBufEmpty 位没被网卡设置的时候，则说明环形缓冲区中有接收到的数据等待处理*/while (netif_running(dev) && received < budget &&(RTL_R8 (ChipCmd) & RxBufEmpty) == 0) {u32 ring_offset = cur_rx % RX_BUF_LEN;u32 rx_status;unsigned int pkt_size;struct sk_buff *skb;rmb();/* 获取接收状态以及接收数据的长度*//* read size+status of next frame from DMA ring buffer */rx_status = le32_to_cpu (*(__le32 *) (rx_ring + ring_offset));rx_size = rx_status >> 16;/* 实际数据包的长度，减去4个字节的CRC*/pkt_size = rx_size - 4;netif_dbg(tp, rx_status, dev, "%s() status %04x, size %04x, cur %04x\n",__func__, rx_status, rx_size, cur_rx);
#if RTL8139_DEBUG > 2print_hex_dump(KERN_DEBUG, "Frame contents: ",DUMP_PREFIX_OFFSET, 16, 1,&rx_ring[ring_offset], 70, true);
#endif/*当EarlyRX 允许的时候，可能会发生这种情况，一个完整的数据包的一部分已经通过DMA 传送到了内存中，而另外一部分还在网卡内部FIFO 中，网卡的DMA 操作还在进行中*//* Packet copy from FIFO still in progress.* Theoretically, this should never happen* since EarlyRx is disabled.*/if (unlikely(rx_size == 0xfff0)) {if (!tp->fifo_copy_timeout)tp->fifo_copy_timeout = jiffies + 2;else if (time_after(jiffies, tp->fifo_copy_timeout)) {netdev_dbg(dev, "hung FIFO. Reset\n");rx_size = 0;goto no_early_rx;}netif_dbg(tp, intr, dev, "fifo copy in progress\n");tp->xstats.early_rx++;break;}no_early_rx:tp->fifo_copy_timeout = 0;/* If Rx err or invalid rx_size/rx_status received* (which happens if we get lost in the ring),* Rx process gets reset, so we abort any further* Rx processing.*/if (unlikely((rx_size > (MAX_ETH_FRAME_SIZE+4)) ||(rx_size < 8) ||(!(rx_status & RxStatusOK)))) {rtl8139_rx_err (rx_status, dev, tp, ioaddr);received = -1;goto out;}/* Malloc up new buffer, compatible with net-2e. *//* Omit the four octet CRC from the length. */skb = netdev_alloc_skb_ip_align(dev, pkt_size);if (likely(skb)) {
#if RX_BUF_IDX == 3wrap_copy(skb, rx_ring, ring_offset+4, pkt_size);
#elseskb_copy_to_linear_data (skb, &rx_ring[ring_offset + 4], pkt_size);
#endifskb_put (skb, pkt_size);skb->protocol = eth_type_trans (skb, dev);dev->stats.rx_bytes += pkt_size;dev->stats.rx_packets++;//数据包从这里进入上层netif_receive_skb (skb);  } else {if (net_ratelimit())netdev_warn(dev, "Memory squeeze, dropping packet\n");dev->stats.rx_dropped++;}received++;/* 前一个4是头部的状态和长度的4个字节，后面的3是为了对齐*/cur_rx = (cur_rx + rx_size + 4 + 3) & ~3;RTL_W16 (RxBufPtr, (u16) (cur_rx - 16));/*清除中断状态位*/rtl8139_isr_ack(tp);}if (unlikely(!received || rx_size == 0xfff0))rtl8139_isr_ack(tp);netdev_dbg(dev, "Done %s(), current %04x BufAddr %04x, free to %04x, Cmd %02x\n",__func__, cur_rx,RTL_R16(RxBufAddr), RTL_R16(RxBufPtr), RTL_R8(ChipCmd));tp->cur_rx = cur_rx;/** The receive buffer should be mostly empty.* Tell NAPI to reenable the Rx irq.*/if (tp->fifo_copy_timeout)received = budget;out:return received;
}

最后就是netif_receive_skb了，数据包从此离开链路层，提交给上层。
六.netif_receive_skb

int netif_receive_skb(struct sk_buff *skb)
{if (netdev_tstamp_prequeue)net_timestamp_check(skb);if (skb_defer_rx_timestamp(skb))return NET_RX_SUCCESS;#ifdef CONFIG_RPS{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();} else {rcu_read_unlock();ret = __netif_receive_skb(skb);}return ret;}
#elsereturn __netif_receive_skb(skb);
#endif
}

netif_receive_skb(skb) 这是一个辅助函数，用于在poll中处理接收到的帧。它主要是向各个已注册的协议处理例程发送一个SKB。

总结一下：
netif_rx是旧的收包函数
比如在某个网卡收到一个包后，首先就是调用这个函数

netif_rx把包放入一个每CPU队列：
__skb_queue_tail(&queue->input_pkt_queue, skb);
并且raise软中断NET_RX_SOFTIRQ，让它进一步处理包，因为收包是在网卡驱动的中断中

最后软中断处理函数 net_rx_action会得到运行，这个函数会对每个收到包的设备调用其设备的出队列函数，把包从上面的队列中拿出来（process_backlog函数），拿出来之后就会调用netif_receive_skb开始靠近协议栈，有很多人可能要处理它，比如PF_PACKET(tcpdump),bridge,等，如果最后包还在，那么就会进入协议栈的3层，对ipv4的包，调用了ipv4的包接收函数，ip_rcv,这个函数在简单的校验之后会到netfilter,如果还幸存，那就复杂了，比如典型的最后就到tcp或者udp的收包程序，它们检查有没有socket需要，不需要就扔掉等。

通俗的讲：
在netif_rx函数中会调用netif_rx_schedule, 然后该函数又会去调用__netif_rx_schedule
在函数__netif_rx_schedule中会去触发软中断NET_RX_SOFTIRQ, 也即是去调用net_rx_action.
然后在net_rx_action函数中会去调用设备的poll函数, 它是设备自己注册的.
在设备的poll函数中, 会去调用neif_receive_skb函数, 在该函数中有下面一条语句 pt_prev->func, 此处的func为一个函数指针, 在之前的注册中设置为ip_rcv.
因此, 就完成了数据包从链路层上传到网络层的这一个过程了.

我们就分析到这里就行了，已经知道驱动程序是如何把数据传送给内核了，至于再往后的操作，这里暂时不管了，以后有机会再分析。

想一起探讨以及获得各种学习资源加我（有我博客中写的代码的原稿）：
qq：1126137994
微信：liu1126137994
可以共同交流关于嵌入式，操作系统，C++语言，C语言，数据结构等技术问题。