当前位置：首页 > news >正文

从零手搓一个简易自旋锁：用atomic_t理解Linux内核并发控制的基石

news 2026/4/29 1:16:32

从零手搓一个简易自旋锁：用atomic_t理解Linux内核并发控制的基石

在Linux内核开发中，并发控制是一个永恒的话题。想象一下，当多个CPU核心同时试图修改同一个内存区域时，如果没有适当的同步机制，数据的一致性将无法保证。这就是为什么我们需要理解像atomic_t这样的基础构建块——它不仅是一个简单的数据类型，更是构建更复杂同步原语的基石。

今天，我们要做的不是简单地调用现成的API，而是从最底层开始，用atomic_t亲手打造一个自旋锁。这个过程就像学习编程时从零开始实现链表一样，虽然内核已经提供了完善的实现，但通过自己动手，我们能获得对并发控制机制更深刻的理解。适合阅读本文的读者包括：正在学习Linux内核开发的初学者，希望深入理解并发原理的中级开发者，或者任何对操作系统底层机制感兴趣的技术爱好者。

1. 理解atomic_t：原子操作的基石

在开始构建自旋锁之前，我们需要先理解atomic_t这个基础数据类型。atomic_t是Linux内核中用于原子操作的特殊整型变量，它的定义通常如下：

typedef struct { int counter; } atomic_t;

看起来简单，但关键在于内核为它提供的一系列原子操作API。这些操作之所以"原子"，是因为它们在执行过程中不会被中断，即使在多核环境下也能保证操作的完整性。常见的原子操作包括：

ATOMIC_INIT(i)：初始化原子变量为值i
atomic_read(v)：读取原子变量v的值
atomic_set(v, i)：设置原子变量v的值为i
atomic_add(i, v)：给v加上i
atomic_sub(i, v)：从v减去i
atomic_inc(v)：v自增1
atomic_dec(v)：v自减1

这些操作在底层通常通过特殊的CPU指令实现，比如x86架构下的LOCK前缀指令，它能确保指令执行期间总线被锁定，防止其他CPU核心干扰。

为什么需要原子操作？考虑一个简单的计数器递增操作counter++，它实际上包含三个步骤：读取值、增加、写回。在多核环境下，两个CPU可能同时读取相同的初始值，各自增加后写回，导致最终结果只增加了一次而非两次。原子操作消除了这种竞态条件。

2. 自旋锁的基本原理

自旋锁(Spinlock)是最简单的同步原语之一，它的行为可以概括为：

当一个线程尝试获取锁时，如果锁已被占用，它会在一个循环中不断检查（"自旋"）直到锁可用
获取锁后执行临界区代码
完成后释放锁

与互斥锁不同，自旋锁不会使线程睡眠，因此适用于以下场景：

临界区代码执行时间非常短
不允许睡眠的上下文（如中断处理程序）
多核系统（单核系统上自旋锁没有意义，因为持有锁的线程无法被抢占释放锁）

自旋锁的核心状态可以用一个简单的整型变量表示：

0：锁未被持有
1：锁已被持有

我们需要实现两个基本操作：

spin_lock：尝试获取锁
spin_unlock：释放锁

3. 实现基础的Test-and-Set操作

要实现自旋锁，最关键的是实现"测试并设置"(Test-and-Set)原子操作。这个操作需要完成以下步骤：

检查锁的当前值
如果为0（未锁定），则设置为1（锁定）并返回成功
如果为1（已锁定），则返回失败

在C语言中，这看起来像：

int test_and_set(int *lock) { if (*lock == 0) { *lock = 1; return 1; // 获取锁成功 } return 0; // 获取锁失败 }

但这段代码不是原子的！我们需要用atomic_t提供的原子操作来实现真正的原子Test-and-Set。Linux内核提供了atomic_cmpxchg函数，它比较并交换值，整个过程是原子的：

// 伪代码，展示原理 int atomic_test_and_set(atomic_t *lock) { int old_value = 0; // 比较lock的值是否为0，如果是则设置为1 // 返回lock原来的值 return atomic_cmpxchg(lock, 0, 1); }

如果返回0，表示我们成功获取了锁；如果返回1，表示锁已被占用。

4. 完整自旋锁实现

现在我们可以基于atomic_t实现完整的自旋锁了。以下是简化版的实现：

#include <linux/atomic.h> typedef struct { atomic_t lock; } spinlock_t; #define SPIN_LOCK_UNLOCKED (spinlock_t){ .lock = ATOMIC_INIT(0) } void spin_lock(spinlock_t *lock) { while (atomic_cmpxchg(&lock->lock, 0, 1) != 0) { // 自旋等待，可以加入CPU放松指令如cpu_relax() ; } } void spin_unlock(spinlock_t *lock) { atomic_set(&lock->lock, 0); }

这个实现虽然简单，但包含了自旋锁的核心逻辑：

spin_lock函数不断尝试用原子操作获取锁，直到成功
spin_unlock函数简单地用原子操作将锁状态重置为0

实际内核实现会更复杂，包括处理锁的公平性、防止过度自旋浪费CPU等问题，但我们的简化版本已经展示了基本原理。

5. 优化与注意事项

基础实现虽然能用，但在实际应用中还需要考虑以下优化点：

5.1 减少缓存竞争

自旋锁在高竞争情况下会导致严重的缓存行颠簸（Cache Line Bouncing），因为所有等待的CPU都在频繁读取锁的状态。可以通过以下方式缓解：

void spin_lock(spinlock_t *lock) { while (atomic_cmpxchg(&lock->lock, 0, 1) != 0) { while (atomic_read(&lock->lock) == 1) // 先读，减少写操作 cpu_relax(); // 提示CPU降低功耗 } }

5.2 防止编译器优化过度

需要使用内存屏障确保编译器和CPU不会对指令重排：

void spin_unlock(spinlock_t *lock) { smp_mb__before_atomic(); // 内存屏障 atomic_set(&lock->lock, 0); }

5.3 自旋锁的使用限制

自旋锁有其适用场景，使用时需注意：

不能递归获取：同一个线程重复获取会导致死锁
临界区必须短小：长时间持有会浪费CPU资源
禁用中断的场景：在某些情况下需要配合spin_lock_irqsave使用

6. 测试我们的自旋锁

为了验证自旋锁的正确性，我们可以编写一个简单的测试模块：

#include <linux/module.h> #include <linux/kthread.h> #include <linux/delay.h> static spinlock_t my_lock = SPIN_LOCK_UNLOCKED; static int shared_counter = 0; static int worker_thread(void *data) { int i; for (i = 0; i < 100000; i++) { spin_lock(&my_lock); shared_counter++; spin_unlock(&my_lock); } return 0; } static int __init spinlock_test_init(void) { struct task_struct *t1, *t2; printk(KERN_INFO "Initializing spinlock test\n"); t1 = kthread_run(worker_thread, NULL, "worker1"); t2 = kthread_run(worker_thread, NULL, "worker2"); msleep(100); // 等待线程完成 printk(KERN_INFO "Final counter value: %d (expected 200000)\n", shared_counter); return 0; } static void __exit spinlock_test_exit(void) { printk(KERN_INFO "Exiting spinlock test\n"); } module_init(spinlock_test_init); module_exit(spinlock_test_exit); MODULE_LICENSE("GPL");

这个测试创建两个线程，每个线程对共享计数器递增10万次。如果没有锁保护，最终结果通常会小于20万；而使用我们的自旋锁后，结果应该是准确的20万。