对于内核处理网卡接受到的帧的流程我是这样理解的:网卡接到一个帧,网卡向处理器发出中断信号,处理器中断正在执行的程序转而执行网卡中断处理程序,将帧拷贝到内核空间,把要执行的下半部程序做上标记,然后打开中断,接著返回。

这里的疑问是,那么对应的下半部程序虽然被标记了,但是怎么执行的呢?是谁在什么时候调起执行的呢?

整个协议栈处理网路数据的过程还是比较需要时间的,所以不可能一直在中断上下文中执行。查看的资料中都提到了放在下半部中执行,可好像都没说下半部是怎么被执行的。另外,中断下半部是在进程上下文中执行的吗?


我将我刚在文章中回答贴在此处以供交流。之所以建议你公开提问,是为了能得到更多相关专业的人的解释和探讨,我作为一个非计算机网路方面的人士,以仅一家之言回答网路相关的问题,怕对你有所误导。下面是我的回答,也希望你可以多多参考别人的回答,以及更多专业方面资料的解释。自己去阅读代码和文档当然是最有效的学习手段,我们每个人的说辞有时都有我们的时间和空间上的局限性。

什么是中断和软中断

首先我们要对中断和软中断有个起码的认识,否则下面的内容就不用看了。如果已经认识的人可跳过此段。

中断,一般指硬体中断,多由系统自身或与之链接的外设(如键盘、滑鼠、网卡等)产生。中断首先是处理器提供的一种响应外设请求的机制,是处理器硬体支持的特性。一个外设通过产生一种电信号通知中断控制器,中断控制器再向处理器发送相应的信号。处理器检测到了这个信号后就会打断自己当前正在做的工作,转而去处理这次中断(所以才叫中断)。当然在转去处理中断和中断返回时都有保护现场和返回现场的操作,这里不赘述,我在下面问题的回答里解释过一些,可以参考:

为什么系统调用时要把一些寄存器保存到内核栈又从内核栈恢复??

www.zhihu.com图标

当然更推荐去阅读更正规的书籍资料来了解中断的进入和返回是如何保护和恢复现场的。不同的设备会对应不同的中断号,不同的中断也会有不同的中断处理函数,中断处理函数一般在设备驱动注册时一同注册,这样一来哪个设备有了事件就能产生对应的中断,并找到对应的中断处理程序来执行了。所以一个硬体中断的大致过程描述是下面这样(非绝对,依具体情况而定,意会一下即可):

+---------+ 产生中断 +----------+ 通知 +-----+
| 硬体设备 | -------------&> | 中断控制器 | -------&> | CPU |
+---------+ +----------+ +-----+
|
V
[中断内核]
|
V
[是否存在中断处理程序?] & irq_exit ----&> 恢复现场 .....

那软中断又是什么呢?我们知道在中断处理时CPU没法处理其它事物,对于网卡来说,如果每次网卡收包时中断的时间都过长,那很可能造成丢包的可能性。当然我们不能完全避免丢包的可能性,以太包的传输是没有100%保证的,所以网路才有协议栈,通过高层的协议来保证连续数据传输的数据完整性(比如在协议发现丢包时要求重传)。但是即使有协议保证,那我们也不能肆无忌惮的使用中断,中断的时间越短越好,尽快放开处理器,让它可以去响应下次中断甚至进行调度工作。基于这样的考虑,我们将中断分成了上下两部分,上半部分就是上面说的中断部分,需要快速及时响应,同时需要越快结束越好。而下半部分就是完成一些可以推后执行的工作。对于网卡收包来说,网卡收到数据包,通知内核数据包到了,中断处理将数据包存入内存这些都是急切需要完成的工作,放到上半部完成。而解析处理数据包的工作则可以放到下半部去执行。

软中断就是下半部使用的一种机制,它通过软体模仿硬体中断的处理过程,但是和硬体没有关系,单纯的通过软体达到一种非同步处理的方式。其它下半部的处理机制还包括tasklet,工作队列等。依据所处理的场合不同,选择不同的机制,网卡收包一般使用软中断。对应NET_RX_SOFTIRQ这个软中断,软中断的类型如下:

enum
{
HI_SOFTIRQ=0,
TIMER_SOFTIRQ,
NET_TX_SOFTIRQ,
NET_RX_SOFTIRQ,
BLOCK_SOFTIRQ,
IRQ_POLL_SOFTIRQ,
TASKLET_SOFTIRQ,
SCHED_SOFTIRQ,
HRTIMER_SOFTIRQ,
RCU_SOFTIRQ, /* Preferable RCU should always be the last softirq */

NR_SOFTIRQS
};

为了代码可查询的准确性,我下面的代码均出自upstream mainline Linux v5.8-rc4,如果其他人在阅读时发现有对不上的函数,请检查自己的内核版本,或者根据理解自行「修补」偏差。

网卡收包的中断过程

  • 注册网卡中断

这其实是一个大问题,我上面说过我不是网路这方面的专业人士,所以不想在此越俎代庖班门弄斧,我下面只针对收包时从中断到软中断的过程进行一定的论述,不再向后续深入讲解。

一个网卡收到数据包,它首先要做的事情就是通知处理器进行中断处理,这一点我们在上面简述中断过程的时候说过了。不同的外设有不同的中断和中断处理函数,所以要研究网卡的中断我们得以一个具体的网卡驱动为例,比如e1000。其模块初始化函数就是:

static int __init e1000_init_module(void)
{
int ret;
pr_info("%s - version %s
", e1000_driver_string, e1000_driver_version);

pr_info("%s
", e1000_copyright);

ret = pci_register_driver(e1000_driver);
...
...
return ret;
}

其中e1000_driver这个结构体是一个关键,它的赋值如下:

static struct pci_driver e1000_driver = {
.name = e1000_driver_name,
.id_table = e1000_pci_tbl,
.probe = e1000_probe,
.remove = e1000_remove,
.driver = {
.pm = e1000_pm_ops,
},
.shutdown = e1000_shutdown,
.err_handler = e1000_err_handler
};

其中很主要的一个方法就是.probe方法,也就是e1000_probe():

/**
* e1000_probe - Device Initialization Routine
* @pdev: PCI device information struct
* @ent: entry in e1000_pci_tbl
*
* Returns 0 on success, negative on failure
*
* e1000_probe initializes an adapter identified by a pci_dev structure.
* The OS initialization, configuring of the adapter private structure,
* and a hardware reset occur.
**/
static int e1000_probe(struct pci_dev *pdev, const struct pci_device_id *ent)
{
...
...
netdev-&>netdev_ops = e1000_netdev_ops;
e1000_set_ethtool_ops(netdev);
...
...
}

这个函数很长,我们不都列出来,这是e1000主要的初始化函数,即使从注释都能看出来。我们留意其注册了netdev的netdev_ops,用的是e1000_netdev_ops这个结构体:

static const struct net_device_ops e1000_netdev_ops = {
.ndo_open = e1000_open,
.ndo_stop = e1000_close,
.ndo_start_xmit = e1000_xmit_frame,
.ndo_set_rx_mode = e1000_set_rx_mode,
.ndo_set_mac_address = e1000_set_mac,
.ndo_tx_timeout = e1000_tx_timeout,
...
...
};

这个e1000的方法集里有一个重要的方法,e1000_open,我们要说的中断的注册就从这里开始:

/**
* e1000_open - Called when a network interface is made active
* @netdev: network interface device structure
*
* Returns 0 on success, negative value on failure
*
* The open entry point is called when a network interface is made
* active by the system (IFF_UP). At this point all resources needed
* for transmit and receive operations are allocated, the interrupt
* handler is registered with the OS, the watchdog task is started,
* and the stack is notified that the interface is ready.
**/
int e1000_open(struct net_device *netdev)
{
struct e1000_adapter *adapter = netdev_priv(netdev);
struct e1000_hw *hw = adapter-&>hw;
...
...
err = e1000_request_irq(adapter);
...
}

e1000在这里注册了中断:

static int e1000_request_irq(struct e1000_adapter *adapter)
{
struct net_device *netdev = adapter-&>netdev;
irq_handler_t handler = e1000_intr;
int irq_flags = IRQF_SHARED;
int err;

err = request_irq(adapter-&>pdev-&>irq, handler, irq_flags, netdev-&>name,
...
...
}

如上所示,这个被注册的中断处理函数,也就是handler,就是e1000_intr()。我们不展开这个中断处理函数看了,我们知道中断处理函数在这里被注册了,在网路包来的时候会触发这个中断函数。

  • 注册软中断

上面我们看到了网卡硬中断的注册,我们下面看一下软中断处理的注册。我们在一开始提到了网卡收包时使用的软中断是NET_RX_SOFTIRQ,我们就在内核中查找这个关键字,看看这个注册的位置在哪。踏破铁鞋无觅处,得来全不费工夫,原来这个注册的位置在这里:

/*
* Initialize the DEV module. At boot time this walks the device list and
* unhooks any devices that fail to initialise (normally hardware not
* present) and leaves us with a valid list of present and active devices.
*
*/
...
static int __init net_dev_init(void)
{
...
...
open_softirq(NET_TX_SOFTIRQ, net_tx_action);
open_softirq(NET_RX_SOFTIRQ, net_rx_action);
...
...
}

open_softirq()函数就是注册软中断用的函数,向它指定软中断号NET_RX_SOFTIRQ和软中断处理函数 net_rx_action()就可以完成注册了。

  • 从硬中断到软中断

现在一个网路包来了,会产生中断,会执行do_IRQ。关于do_IRQ的实现有很多,不同硬体对中断的处理都会有所不同,但一个基本的执行思路就是:

void __irq_entry do_IRQ(unsigned int irq) |do_IRQ[98] void do_IRQ(struct pt_regs *regs, int irq)
{ |
irq_enter(); |*** arch/sh/kernel/irq.c: |do_IRQ[185] asmlinkage __irq_entry int do_IRQ(unsigned int irq, struct pt_r
generic_handle_irq(irq); |egs *regs)
irq_exit(); |
}

我们没必要都展开,让我们专注我们的问题。do_IRQ会执行上面e1000_intr这个中断处理函数,这个中断处理是属于上半部的处理,在do_IRQ的结尾会调用irq_exit(),这是软中断和中断衔接的一个地方。我们重点说一下这里。

void irq_exit(void)
{
__irq_exit_rcu();
rcu_irq_exit();
/* must be last! */
lockdep_hardirq_exit();
}

static inline void __irq_exit_rcu(void)
{
#ifndef __ARCH_IRQ_EXIT_IRQS_DISABLED
local_irq_disable();
#else
lockdep_assert_irqs_disabled();
#endif
account_irq_exit_time(current);
preempt_count_sub(HARDIRQ_OFFSET);
if (!in_interrupt() local_softirq_pending())
invoke_softirq();

tick_irq_exit();
}

在irq_exit()的第一步就是一个local_irq_disable(),也就是说禁止了中断,不再响应中断。因为下面要处理所有标记为要处理的软中断,关中断是因为后面要清除这些软中断,将CPU软中断的点阵图中置位的位清零,这需要关中断,防止其它进程对点阵图的修改造成干扰。

然后preempt_count_sub(HARDIRQ_OFFSET),硬中断的计数减1,表示当前的硬中断到这里就结束了。但是如果当前的中断是嵌套在其它中断里的话,这次减1后不会计数清0,如果当前只有这一个中断的话,这次减1后计数会清0。注意这很重要。

因为接下来一步判断!in_interrupt() local_softirq_pending(),第一个!in_interrupt()就是通过计数来判断当前是否还处于中断上下文中,如果当前还有为完成的中断,则直接退出当前中断。后半部的执行在后续适当的时机再进行,这个「适当的时机」比如ksoftirqd守护进程的调度,或者下次中断到此正好不在中断上下文的时候等情况。

我们现在假设当前中断结束后没有其它中断了,也就是不在中断上下文了,且当前CPU有等待处理的软中断,即local_softirq_pending()也为真。那么执行invoke_softirq()。

static inline void invoke_softirq(void)
{
if (ksoftirqd_running(local_softirq_pending()))
return;

if (!force_irqthreads) {
#ifdef CONFIG_HAVE_IRQ_EXIT_ON_IRQ_STACK
/*
* We can safely execute softirq on the current stack if
* it is the irq stack, because it should be near empty
* at this stage.
*/
__do_softirq();
#else
/*
* Otherwise, irq_exit() is called on the task stack that can
* be potentially deep already. So call softirq in its own stack
* to prevent from any overrun.
*/
do_softirq_own_stack();
#endif
} else {
wakeup_softirqd();
}
}

这个函数的逻辑很简单,首先如果ksoftirqd正在被执行,那么我们不想处理被pending的软中断,交给ksoftirqd线程来处理,这里直接退出。

如果ksoftirqd没有正在运行,那么判断force_irqthreads,也就是判断是否配置了CONFIG_IRQ_FORCED_THREADING,是否要求强制将软中断处理都交给ksoftirqd线程。因为这里明显要在中断处理退出的最后阶段处理软中断,但是也可以让ksoftirqd来后续处理。如果设置了force_irqthreads,则不再执行__do_softirq(),转而执行wakeup_softirqd()来唤醒ksoftirqd线程,将其加入可运行队列,然后退出。

如果没有设置force_irqthreads,那么就执行__do_softirq():

asmlinkage __visible void __softirq_entry __do_softirq(void)
{
...
...
pending = local_softirq_pending();
account_irq_enter_time(current);

__local_bh_disable_ip(_RET_IP_, SOFTIRQ_OFFSET);
in_hardirq = lockdep_softirq_start();

restart:
/* Reset the pending bitmask before enabling irqs */
set_softirq_pending(0);

local_irq_enable();

h = softirq_vec;

while ((softirq_bit = ffs(pending))) {
...
...
}

if (__this_cpu_read(ksoftirqd) == current)
rcu_softirq_qs();
local_irq_disable();

pending = local_softirq_pending();
if (pending) {
if (time_before(jiffies, end) !need_resched()
--max_restart)
goto restart;

wakeup_softirqd();
}

lockdep_softirq_end(in_hardirq);
account_irq_exit_time(current);
__local_bh_enable(SOFTIRQ_OFFSET);
WARN_ON_ONCE(in_interrupt());
current_restore_flags(old_flags, PF_MEMALLOC);
}

注意在函数开始时就先执行了一个__local_bh_disable_ip(_RET_IP_, SOFTIRQ_OFFSET),表示当前要处理软中断了,在这种情况下是不允许睡眠的,也就是不能进程调度。这点很重要,也很容易混淆,加上前面我们说的irq_exit()开头的local_irq_disable(),所以当前处在一个既禁止硬中断,又禁止软中断,不能睡眠不能调度的状态。很多人就容易将这种状态归类为「中断上下文」,我个人认为是不对的。从上面in_interrupt函数的定义来看,是否处于中断上下文和preempt_count对于中断的计数有关:

#define irq_count() (preempt_count() (HARDIRQ_MASK | SOFTIRQ_MASK
| NMI_MASK))

#define in_interrupt() (irq_count())

和是否禁止了中断没有直接的关系。虽然中断上下文应该不允许睡眠和调度,但是不能睡眠和调度的时候不等于in_interrupt,比如spin_lock的时候也是不能睡眠的(这是目前我个人观点)。但是很多程序员之所以容易一概而论,是因为对于内核程序员来讲,判断自己所编程的位置是否可以睡眠和调度是最被关心的,所以禁用了中断后不能调度和睡眠就很容易被归类为在中断上下文,实际上我个人认为这应该算一个误解,或者说是「变相扩展」后的说辞。一切还要看我们对中断上下文这个概念的界定,如果像in_interrupt那样界定,那关不关中断和是否处于中断上下文就没有直接的关系。

下面在__do_softirq开始处理软中断(执行每一个待处理的软中断的action)前还有一个很关键的地方,就是local_irq_enable(),这就打开了硬体中断,然后后面的软中断处理可以在允许中断的情况下执行。注意这时候__local_bh_disable_ip(_RET_IP_, SOFTIRQ_OFFSET)仍然有效,睡眠仍然是不允许的。

到这里我们可以看到,内核是尽量做到能允许中断就尽量允许,能允许调度就尽量允许,因为无情的禁止是对CPU资源最大的浪费,也是对外设中断的不负责。否则长期处于禁止中断的情况下,网卡大量丢包将是难免的,而这也将是制约成网卡实际速率的瓶颈。

结语

因为提问者一直在纠结网卡收包时软中断和中断的关系,以及软中断到底是不是在中断上下文这种问题。所以我围绕此重点讲述,更多其它的内容就不涉及了。再次强调我不是这方面的专业人士,上述内容仅凭我个人短时间的代码阅读理解,纯属越俎代庖,如有网路方面的专业人士路过看到有明显疏漏的地方还请不吝指出。


内容来自SJTU,IPADS OS-16-Network

Linux全流程

既然要讲,那就把一个包的整个包生都说了算了

触发中断

  • 在非虚拟化环境下,网卡通过DMA将packet写入内核的rx_ring环形队列缓冲区,并触发中断。
  • 如果在虚拟化环境下,VMM配置GIC ITS (Interrupt Translation Service) ,建立物理中断与虚拟中断的映射完成中断虚拟化使得网卡能直接向VM发出中断,同时通过IO虚拟化,网卡通过IOMMU将packet直接写入虚拟机内核的rx_ring

Top Half

  • CPU在收到中断之后,调用网卡ISR也就是所谓的中断handler
  • 分配sk_buf并入input_pkt_queue(如果队列已满则丢弃)
  • 发出一个软中断NET_RX_SOFTIRQ,软中断可以被调度例如通过tasklet

Bottom Half

  • sk_buf从input_pkt_queue传入process_queue,根据协议类型调用网路层协议的handler
  • ip_rcv执行包头检查,ip_router_input()进行路由,决定本机/转发/丢弃
  • tcp_v4_rcv执行包头检查,tcp_v4_lookup查询对应的socket和connection,如果正常,tcp_prequeue将skb放进socket接收队列
  • socket随即唤醒所在的进程

Kqueue

因为epoll没有论文,就说说kqueue是怎么做的吧,kqueue会根据socket绑定的knote链表(每个监听的kqueue都可能创建一个knote),将knote通过反向指针获得kqueue,将knote加入kqueue的就绪队列末尾。如果此时恰好有进程正在监听的话,将会唤醒进程,kqueue会被扫描,并从就绪队列处获得所有的event,从而了解已经就绪的所有socket。

https://zhuanlan.zhihu.com/p/157431765?

zhuanlan.zhihu.com图标
  • 唤醒的进程调用socket recv系统调用,如果是TCP则调用tcp_recvmsg从sk_buffer拷贝数据

Batch

netif_receive_skb_list()

Linux的NAPI还会继续延迟软中断的处理,等待其积累足够的skb后进行轮询,一次性处理所有的skb。

SKB

skb并不是直接存储报文,而是存储指针,指针只需要移动,就能完成解包,而本身的报文并不需要修改。上一层的协议栈会在处理当前层的同时设置好下一层的头指针,并且移动data指针。与此同时,skb本身是双向链表实现的队列。qlen为链表元素长度,lock为添加元素时的锁。


谈到指针的用法,这里举个做OS lab时印象深刻的奇淫巧技,也是C的指针变态的地方

#define list_entry(ptr, type, field)
container_of(ptr, type, field)
#define container_of(ptr, type, field)
((type *)((void *)(ptr) - (u64)((((type *)(0))-&>field))))

(u64)((((type *)(0))-&>field))))指的是field在结构体type中的偏移量,通过减去这个偏移量我们就能找出某个对象所在上级type对象的地址,也就是container。

一般来说,我们都会使用下面这样的方式,让链表节点去包裹数据。

struct page_list_node {
struct page* p;
struct page_list_node *prev;
struct page_list_node *next;
};

但是,通过指针操作,却可以让数据去包裹链表节点

struct list_head {
struct list_head *prev;
struct list_head *next;
};

struct page{
struct list_head list_node;
}

在仅仅知道链表节点的情况下,借助成员偏移量即可知道容器对象的位置并取出

list_entry(somenode,struct page,list_node);

感谢评论区分享的侵入性容器概念。这个东西是linux的经典宏,同时也是C++ boost的改进。

侵入性的好处在于,元素本身就是iterator,仅仅需要元素本身就能够从容器中删除或迁移到别的容器,不需要关注存在于哪个容器中。

数据结构就是数据结构,和数据分离考虑。因此不需要为各种链表分别按照数据类型创建list类型,只需要解释的时候附带类型就可。


硬体中断分上下2部分,上半部分只通知有中断发生,设置标记,就会打开中断。下半部分叫软中断。

第一次硬体中断,会触发软中断。

软中断是开中断的,因此软中断运行过程中会发生硬体中断。这时就属于硬体中断重入。重入之后不会执行新的软中断,因为第一次的时候软中断已经在执行了。

重入的硬体中断处理会马上退出。

而第一次进入的软中断在处理完之后,会继续判断是否有软中断发生(即发生了硬中断重入),如果有会继续执行。

但是软中断不会一直执行,有个时间判断,超时了就会终止执行,启动一个内核线程来继续执行软中断。

比如当前是用户线程A,发生硬体中断,那么软中断执行的时候,当前线程还是A,但是执行时间久了之后,会卡住系统,因为软中断是除了硬体中断之外优先顺序最高的,软中断执行过程中,整个系统就调度就停止了。为了避免卡住系统,处理时间久了之后(2ms),就会启动一个内核线程B(tasklet),专门处理软中断,而A线程会继续,从内核态退出。

软中断之后有的就会结束这次硬体中断逻辑。有的还会启动workqueue继续处理。比如网路协议栈,除了从设备读写之外,还要继续处理协议内容。因为软中断优先顺序太高,不能执行太长时间,需要长时间处理的,会再分级别。不要紧的部分放到workqueue线程去处理。


STM32是通过中断,就和其他汇流排外设一样。别的硬体我不知道


推荐阅读:
相关文章