[TOC]

include/linux/preempt.h 内核抢占(Kernel Preemption) 控制内核代码的可抢占性与延迟

历史与背景

这项技术是为了解决什么特定问题而诞生的？

include/linux/preempt.h 是定义**内核抢占（Kernel Preemption）相关API和配置的头文件。这项技术的诞生是为了解决早期Linux内核的一个核心设计局限性：内核态执行的不可抢占性（non-preemptibility），以及由此导致的系统响应延迟（latency）**问题。

在Linux 2.6版本之前，内核是非抢占式的。这意味着，当一个进程通过系统调用进入内核态执行时，它会一直占有CPU，直到它自愿放弃（例如，因等待I/O而睡眠）或执行完毕返回用户空间。这种模型的缺点是：

  1. 高延迟：如果一个低优先级的进程执行了一个非常耗时的系统调用（例如，对一个大文件进行复杂的读写），那么一个刚刚被唤醒的、需要立即响应用户输入的高优先级进程（如桌面窗口管理器或文本编辑器）将不得不一直等待，直到那个系统调用完成。这会导致系统UI卡顿，响应迟钝。
  2. 实时性差：对于实时系统，这种不可预测的、长时间的延迟是致命的。

内核抢占机制的引入，就是为了改变这种状况。它允许在一个进程正在内核态执行时，如果一个更高优先级的进程变为可运行状态，调度器可以立即中断当前进程，抢占CPU并将其分配给那个更高优先级的进程。preempt.h 提供了实现这一机制所需的底层构建块。

它的发展经历了哪些重要的里程碑或版本迭代？

Linux的可抢占性是一个分阶段演进的、不断增强的过程。

  • 用户抢占（User Preemption）：这是Linux一直具备的能力。当内核从一个中断或系统调用返回用户空间时，它会检查是否有更高优先级的任务需要运行，并进行抢占。
  • 内核抢占（Kernel Preemption, CONFIG_PREEMPT）：这是在2.5/2.6开发周期中的一个革命性里程碑。它使得内核在从中断处理程序返回内核空间时，如果发现有更高优先级的任务就绪，也可以进行抢占。但是，这种抢占是有条件的：它只在内核代码**没有显式持有锁（如自旋锁）**时才发生。
  • 完全实时抢占（PREEMPT_RT）：这是内核抢占的终极形态。由实时Linux补丁集（PREEMPT_RT patchset）引入，现在绝大部分已合入主线内核。它通过将内核中大部分的自旋锁替换为可睡眠的、支持优先级继承的互斥锁，使得内核在几乎任何地方（除了极少数真正的临界区）都是可抢占的。这极大地降低了内核内部的调度延迟，是构建硬实时Linux系统的基础。

目前该技术的社区活跃度和主流应用情况如何？

内核抢占是现代Linux内核的一个基础特性。

  • 主流应用：
    • 桌面系统：几乎所有桌面发行版都会开启CONFIG_PREEMPT_DESKTOP，以获得流畅的用户体验。
    • 服务器：通常会选择CONFIG_PREEMPT_VOLUNTARY或完全不抢占（CONFIG_PREEMPT_NONE），以追求更高的吞吐量。
    • 实时系统：工业控制、电信、金融等领域广泛使用开启了PREEMPT_RT的Linux内核。
      preempt.h 中定义的API是所有这些配置下，内核代码正确同步的基础。

核心原理与设计

它的核心工作原理是什么？

内核抢占的核心是围绕一个名为**preempt_count的per-thread（每个线程独立）计数器**来工作的。这个计数器像一个“请勿打扰”的标志。

  1. preempt_count计数器：

    • 这是一个32位的整数，其不同的位段被用于跟踪不同的状态（是否在中断中、是否持有自旋锁等），但其核心是抢占计数值。
    • 当一个线程进入一个不可被抢占的区域时，它会调用preempt_disable()，这个函数会增加preempt_count的值。
    • 当它离开这个区域时，会调用preempt_enable()，减少preempt_count的值。

  2. 抢占的条件：

    • 一个在内核态运行的线程是可抢占的，当且仅当其preempt_count的值为0。

  3. 抢占的触发点：

    • 抢占并不是随时随地发生的。调度器只在特定的检查点检查是否需要抢占。最主要的检查点是：
      • 从一个硬件中断处理程序返回内核空间时。
      • 当代码显式地调用preempt_enable()，并且在调用后preempt_count恰好变为0时。
    • 在这些检查点，如果内核发现当前任务的TIF_NEED_RESCHED标志被设置了（意味着有更高优先级的任务在等待），并且preempt_count为0，那么抢占就会发生。

  4. preempt.h 提供的API：

    • preempt_disable() / preempt_enable()：控制抢占计数的核心API。
    • preempt_count()：读取当前preempt_count的值。
    • in_atomic() / in_interrupt() / in_serving_softirq()：这些是极其常用的宏，它们通过检查preempt_count的不同位段来判断当前代码是否处于一个原子上下文（即不可睡眠的上下文）。这是开发者用来判断自己是否可以调用睡眠函数的标准方式。

它的主要优势体现在哪些方面？

  • 降低系统延迟：极大地提高了交互式应用的响应速度。
  • 支持实时性：是构建实时Linux系统的基础。
  • 提供了精细的控制：允许内核开发者通过preempt_disable/enable来精确地界定那些绝对不能被打断的、极短的临界区。

它存在哪些已知的劣势、局限性或在特定场景下的不适用性？

  • 性能开销：抢占检查和preempt_count的维护会给内核带来微小的、但不可忽略的性能开销。对于纯粹追求计算吞吐量的HPC场景，关闭内核抢占可以获得轻微的性能提升。
  • 增加了并发复杂性：内核抢占意味着内核代码中可能出现更多的并发路径。一段原本被认为是“原子”执行的代码路径，现在可能会被另一个任务中断，因此需要开发者使用更审慎的锁策略。

使用场景

在哪些具体的业务或技术场景下，它是首选解决方案？

preempt.h中的API主要由内核开发者在编写底层代码时使用，而不是由用户空间程序直接调用。

  • 保护Per-CPU数据：当一段代码需要访问一个per-CPU变量，并且不希望在访问过程中被调度到另一个CPU上（这会导致访问了错误的per-CPU变量）时，一种常见的做法是使用preempt_disable()和preempt_enable()将这段代码包裹起来。因为它只在当前CPU执行，所以不需要昂贵的自旋锁。
  • 实现同步原语：几乎所有的自旋锁实现，在获取锁时都会隐式地调用preempt_disable()，在释放锁时调用preempt_enable()。这是为了防止在持有锁时被抢占，从而导致其他试图获取该锁的CPU长时间自旋，造成死锁。
  • 上下文检查：内核代码在调用一个可能睡眠的函数（如kmalloc(GFP_KERNEL)）之前，必须检查自己是否处于原子上下文。BUG_ON(in_atomic()); 是一种常见的防御性编程。

是否有不推荐使用该技术的场景？为什么？

  • 保护跨CPU共享数据：preempt_disable()绝对不能被用来保护被多个CPU共享的数据。它只阻止了本地CPU上的任务切换，无法阻止另一个CPU上的任务同时访问该数据。这种场景必须使用自旋锁或互斥锁。
  • 长时间的临界区：preempt_disable()应该只用于保护非常短的代码路径。长时间地禁用抢占会严重损害系统延迟，使其效果适得其反。

对比分析

请将其 与 其他相似技术 进行详细对比。

preempt_disable是内核中用于控制并发的多种底层工具之一。

特性 preempt_disable() / enable() spin_lock() / unlock() local_irq_disable() / enable()
保护对象 本地CPU上的调度。 多个CPU对共享数据的互斥访问。 本地CPU上的中断处理。
主要作用 防止当前任务被其他任务抢占。 防止其他CPU同时进入临界区。 防止中断处理程序打断当前代码的执行。
并发模型 阻止任务并发 (在本地CPU)。 阻止CPU并发 (在所有CPU)。 阻止代码与中断的并发 (在本地CPU)。
开销 低 (通常是原子增减和内存屏障)。 中等 (涉及锁总线和缓存一致性协议)。 低 (通常是一条CPU指令)。
副作用 允许中断继续发生和处理。 隐式地禁用本地CPU抢占。 隐式地禁用本地CPU抢占。
适用场景 保护per-CPU数据，防止在访问过程中被迁移。 保护全局或跨CPU共享的、在原子上下文中访问的数据。 保护需要与中断处理程序同步的、极其短暂的临界区。

include/asm-generic/preempt.h

preempt_count 抢占计数

1
2
3
4
static __always_inline int preempt_count(void)
{
	return READ_ONCE(current_thread_info()->preempt_count);
}

__preempt_count_add __preempt_count_sub 增加 减少 cpu的抢占计数

  • preempt_count_ptr() 的目的是返回 preempt_count 的地址，而不是直接读取其值。
    • 返回指针本身不涉及数据读取，因此不需要使用 READ_ONCE。
  • 指针操作的目的：
    • 调用者通过返回的指针可以直接操作 preempt_count 的值，例如读取或修改。
      数据一致性和原子性需要在使用指针时由调用者负责，而不是在返回指针时处理。
  • 避免额外开销：
    • READ_ONCE 是为防止编译器优化和重排序而设计的，但在返回指针的场景中，这种保护是多余的，因为指针本身不会被并发修改
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
static __always_inline volatile int *preempt_count_ptr(void)
{
	return &current_thread_info()->preempt_count;
}
/*
 * The various preempt_count add/sub methods
 */

static __always_inline void __preempt_count_add(int val)
{
	*preempt_count_ptr() += val;
}

static __always_inline void __preempt_count_sub(int val)
{
	*preempt_count_ptr() -= val;
}

//检查当前cpu的抢占计数下溢
//检查抢占计数即将上溢
void preempt_count_add(int val)
{
#ifdef CONFIG_DEBUG_PREEMPT
	/*
	 * Underflow?
	 */
	if (DEBUG_LOCKS_WARN_ON((preempt_count() < 0)))
		return;
#endif
	__preempt_count_add(val);
#ifdef CONFIG_DEBUG_PREEMPT
	/*
	 * Spinlock count overflowing soon?
	 */
	DEBUG_LOCKS_WARN_ON((preempt_count() & PREEMPT_MASK) >=
				PREEMPT_MASK - 10);
#endif
	preempt_latency_start(val);
}
EXPORT_SYMBOL(preempt_count_add);
NOKPROBE_SYMBOL(preempt_count_add);

void preempt_count_sub(int val)
{
#ifdef CONFIG_DEBUG_PREEMPT
	/*
	 * Underflow?
	 */
	if (DEBUG_LOCKS_WARN_ON(val > preempt_count()))
		return;
	/*
	 * Is the spinlock portion underflowing?
	 */
	if (DEBUG_LOCKS_WARN_ON((val < PREEMPT_MASK) &&
			!(preempt_count() & PREEMPT_MASK)))
		return;
#endif

	preempt_latency_stop(val);
	__preempt_count_sub(val);
}
EXPORT_SYMBOL(preempt_count_sub);
NOKPROBE_SYMBOL(preempt_count_sub);

__preempt_count_dec_and_test 抢占计数减少和测试

  • 返回是否可以调度.计数减少到0且允许调度返回1
1
2
3
4
5
6
7
static __always_inline bool __preempt_count_dec_and_test(void)
{
	/* 由于 load-store 架构无法执行每 cpu 的原子作;我们不能使用 PREEMPT_NEED_RESCHED因为它可能会丢失。
	 */
	return !--*preempt_count_ptr() 	//抢占计数减少
	&& tif_need_resched();			//设置需要调度bit位
}

preempt_count_ptr 获取抢占计数指针

1
2
3
4
static __always_inline volatile int *preempt_count_ptr(void)
{
	return &current_thread_info()->preempt_count;
}

preempt_count_dec_and_test 抢占计数减少并测试

  • 函数用于减少抢占计数并测试是否需要进行调度
1
2
3
4
5
6
7
#define preempt_count_dec_and_test() __preempt_count_dec_and_test()

static __always_inline bool __preempt_count_dec_and_test(void)
{
	/*	 * 由于负载-存储架构无法进行每个 CPU 的原子操作；我们无法使用 PREEMPT_NEED_RESCHED，因为它可能会丢失。	 */
	return !--*preempt_count_ptr() && tif_need_resched();
}

include/linux/preempt.h

preemption counter 抢占计数标识

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
/*
 * 我们将 hardirq 和 softirq 计数器放入抢占中
 *计数器。位掩码具有以下含义：
 *
 * - 位 0-7 是抢占计数（最大抢占深度：256）
 * - 第 8-15 位是软中断计数（软中断的最大 # 值：256）
 *
 * 理论上 hardirq 计数可以与
 * 中断，但我们使用
 * interrupts disabled，因此我们不能有嵌套中断。虽然
 * 有一些 Palaeontologic 驱动程序可以重新启用
 * 处理程序，所以我们在这里需要不止一个位。
 *
 * PREEMPT_MASK：0x000000ff
 * SOFTIRQ_MASK：0x0000ff00
 * HARDIRQ_MASK：0x000f0000
 * NMI_MASK：0x00f00000
 * PREEMPT_NEED_RESCHED：0x80000000
 */
 #define PREEMPT_BITS	8
#define SOFTIRQ_BITS	8
#define HARDIRQ_BITS	4
#define NMI_BITS	4

#define PREEMPT_SHIFT	0
#define SOFTIRQ_SHIFT	(PREEMPT_SHIFT + PREEMPT_BITS)
#define HARDIRQ_SHIFT	(SOFTIRQ_SHIFT + SOFTIRQ_BITS)
#define NMI_SHIFT	(HARDIRQ_SHIFT + HARDIRQ_BITS)

#define __IRQ_MASK(x)	((1UL << (x))-1)

#define PREEMPT_MASK	(__IRQ_MASK(PREEMPT_BITS) << PREEMPT_SHIFT)
#define SOFTIRQ_MASK	(__IRQ_MASK(SOFTIRQ_BITS) << SOFTIRQ_SHIFT)
#define HARDIRQ_MASK	(__IRQ_MASK(HARDIRQ_BITS) << HARDIRQ_SHIFT)
#define NMI_MASK	(__IRQ_MASK(NMI_BITS)     << NMI_SHIFT)

#define PREEMPT_OFFSET	(1UL << PREEMPT_SHIFT)
#define SOFTIRQ_OFFSET	(1UL << SOFTIRQ_SHIFT)
#define HARDIRQ_OFFSET	(1UL << HARDIRQ_SHIFT)
#define NMI_OFFSET	(1UL << NMI_SHIFT)

#define SOFTIRQ_DISABLE_OFFSET	(2 * SOFTIRQ_OFFSET)

#define PREEMPT_DISABLED	(PREEMPT_DISABLE_OFFSET + PREEMPT_ENABLED)

DEFINE_LOCK_GUARD_0(preempt)

1
DEFINE_LOCK_GUARD_0(preempt, preempt_disable(), preempt_enable())

PERCPU_PTR 获取percpu指针

  • PERCPU_PTR 是一个宏，用于将传入的指针转换为 percpu 指针类型。它使用了 __force 属性来强制转换类型，以确保编译器不会对类型进行不必要的检查。
  • typeof(*(__p)) 是一个 GCC 扩展，用于获取指针 __p 指向的类型。这个表达式的目的是获取指针所指向的数据类型。
  • __kernel 是一个宏，通常用于指示内核空间的类型。它可能是一个特定于架构的宏，用于标识内核空间的数据类型。
1
2
#define PERCPU_PTR(__p)							\
	(typeof(*(__p)) __force __kernel *)((__force unsigned long)(__p))

per_cpu_ptr raw_cpu_ptr this_cpu_ptr 获取percpu指针

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
#ifdef CONFIG_SMP

/*
 * Add an offset to a pointer.  Use RELOC_HIDE() to prevent the compiler
 * from making incorrect assumptions about the pointer value.
 */
#define SHIFT_PERCPU_PTR(__p, __offset)					\
	RELOC_HIDE(PERCPU_PTR(__p), (__offset))

#define per_cpu_ptr(ptr, cpu)						\
({									\
	__verify_pcpu_ptr(ptr);						\
	SHIFT_PERCPU_PTR((ptr), per_cpu_offset((cpu)));			\
})

#define raw_cpu_ptr(ptr)						\
({									\
	__verify_pcpu_ptr(ptr);						\
	arch_raw_cpu_ptr(ptr);						\
})

#ifdef CONFIG_DEBUG_PREEMPT
#define this_cpu_ptr(ptr)						\
({									\
	__verify_pcpu_ptr(ptr);						\
	SHIFT_PERCPU_PTR(ptr, my_cpu_offset);				\
})
#else
#define this_cpu_ptr(ptr) raw_cpu_ptr(ptr)
#endif

#else	/* CONFIG_SMP */

#define per_cpu_ptr(ptr, cpu)						\
({									\
	(void)(cpu);							\
	__verify_pcpu_ptr(ptr);						\
	PERCPU_PTR(ptr);						\
})

#define raw_cpu_ptr(ptr)	per_cpu_ptr(ptr, 0)
#define this_cpu_ptr(ptr)	raw_cpu_ptr(ptr)

#endif	/* CONFIG_SMP */

nmi_count hardirq_count softirq_count irq_count 获取中断计数

1
2
3
4
5
6
7
8
9
10
11
12
/*
 * 这些宏定义避免了 preempt_count（） 的冗余调用，因为鉴于 preempt_count（） 通常使用 READ_ONCE（） 实现，此类调用会导致冗余加载。
 */
#define nmi_count()	(preempt_count() & NMI_MASK)
#define hardirq_count()	(preempt_count() & HARDIRQ_MASK)
#ifdef CONFIG_PREEMPT_RT
# define softirq_count()	(current->softirq_disable_cnt & SOFTIRQ_MASK)
# define irq_count()		((preempt_count() & (NMI_MASK | HARDIRQ_MASK)) | softirq_count())
#else
# define softirq_count()	(preempt_count() & SOFTIRQ_MASK)
# define irq_count()		(preempt_count() & (NMI_MASK | HARDIRQ_MASK | SOFTIRQ_MASK))
#endif

in_irq in_softirq in_interrupt 检查中断状态

1
2
3
4
5
6
7
8
9
/*
 * 以下宏已弃用，不应在新代码中使用：
 * in_irq（） - in_hardirq（） 的过时版本
 * in_softirq（） - 我们禁用了 BH，或者正在处理软中断
 * in_interrupt（） - 我们处于 NMI、IRQ、SoftIRQ 上下文中或禁用了 BH
 */
#define in_irq()		(hardirq_count())
#define in_softirq()		(softirq_count())
#define in_interrupt()		(irq_count())

preempt_disable 禁用抢占

1
2
3
4
5
#define preempt_disable() \
do { \
	preempt_count_inc(); \
	barrier(); \
} while (0)

preempt_enable 启用抢占

  • 在启用抢占时，检查是否需要进行调度
  • 如果需要调度，则调用 __preempt_schedule() 函数
  • 如果不需要调度，则仅仅减少抢占计数
  • 当计数为 0 时，表示可以进行抢占,从而执行调度
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#ifdef CONFIG_PREEMPTION
#define preempt_enable() \
do { \
	barrier(); \
	if (unlikely(preempt_count_dec_and_test())) \
		__preempt_schedule(); \
} while (0)

#else /* !CONFIG_PREEMPTION */
#define preempt_enable() \
do { \
	barrier(); \
	preempt_count_dec(); \
} while (0)
#endif /* CONFIG_PREEMPTION */

preemptible 检查是否可抢占

  • 检查当前的抢占计数是否为 0，并且中断是否未被禁用
  • 如果满足这两个条件，则表示当前任务是可抢占的
1
#define preemptible()	(preempt_count() == 0 && !irqs_disabled())

preempt_disable_notrace 禁用抢占并停止追踪

1
2
3
4
5
#define preempt_disable_notrace() \
do { \
	__preempt_count_inc(); \
	barrier(); \
} while (0)

preempt_enable_no_resched_notrace 启用调度且不需要调度

1
2
3
4
5
#define preempt_enable_no_resched_notrace() \
do { \
	barrier(); \
	__preempt_count_dec(); \
} while (0)

kernel/sched/core.c

schedule_preempt_disabled 在禁用抢占的情况下调用

1
2
3
4
5
6
7
8
9
10
11
/**
 * schedule_preempt_disabled - 在禁用抢占的情况下调用
 *
 * 禁用抢占的情况下返回。注意：preempt_count 必须为 1
 */
void __sched schedule_preempt_disabled(void)
{
	sched_preempt_enable_no_resched();
	schedule();
	preempt_disable();
}

__preempt_schedule 执行调度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/*
 * This is the entry point to schedule() from in-kernel preemption
 * off of preempt_enable.
 */
asmlinkage __visible void __sched notrace preempt_schedule(void)
{
	/* 如果 preempt_count 不为零或中断被禁用，	 
    * 我们不希望抢占当前任务。就直接返回。	 */
	if (likely(!preemptible()))
		return;
	preempt_schedule_common();
}
NOKPROBE_SYMBOL(preempt_schedule);
EXPORT_SYMBOL(preempt_schedule);

preempt_schedule_common 通用的抢占调度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
/*
 * __sched: 告诉编译器这个函数会调用调度器，可能会导致上下文切换。
 * notrace: 告诉内核追踪器(ftrace)，不要追踪这个函数的入口和出口。
 *          这是为了防止无限递归，因为本函数自身就是抢占路径的一部分。
 */
static void __sched notrace preempt_schedule_common(void)
{
	/*
	 * 使用do-while循环，确保在退出前，所有的抢占请求都被处理完毕。
	 */
	do {
		/*
		 * 注释解释了一个复杂的问题：
		 * 因为函数追踪器(ftrace)可能会追踪preempt_count_sub()这类函数，
		 * 而ftrace本身也需要调用preempt_enable/disable_notrace()来保护自己。
		 * 如果NEED_RESCHED标志被设置，那么ftrace调用的preempt_enable_notrace()
		 * 可能会再次调用本函数，从而导致无限递归。
		 *
		 * 为了解决这个问题，必须在ftrace可能开始追踪之前，就先禁用抢占。
		 * 因此，将preempt_disable()拆分为两个调用：
		 * 1. preempt_disable_notrace(): 先禁用抢占，这个操作本身不会被追踪。
		 * 2. preempt_latency_start(): 记录抢占延迟，这个操作可以被追踪。
		 */

		/* 增加抢占计数值，禁用抢占，并且这个操作本身不被ftrace追踪。*/
		preempt_disable_notrace();
		/* 开始记录抢占延迟，用于实时性分析。参数1表示这是一个抢占延迟。*/
		preempt_latency_start(1);
		
		/*
		 * 调用核心调度函数__schedule()。
		 * SM_PREEMPT参数告诉调度器，这是一次由内核抢占触发的调度。
		 * 这个函数执行完毕返回时，CPU可能已经运行了其他任务，然后才切换回来。
		 */
		__schedule(SM_PREEMPT);
		
		/* 停止记录抢占延迟。*/
		preempt_latency_stop(1);
		/*
		 * 减少抢占计数值，但不检查调度请求。
		 * 因为我们马上就要在while循环中检查了，这里不能再次触发调度。
		 * 同样，这个操作本身不被ftrace追踪。
		 */
		preempt_enable_no_resched_notrace();

		/*
		 * 注释：再次检查，以防在schedule返回和现在之间，我们错过了
		 *       一次抢占机会。
		 */
	} while (need_resched()); /* 只要need_resched()为真，就一直循环。*/
	/* static __always_inline bool need_resched(void)
		{
			return unlikely(tif_need_resched());
		}  
	*/
}