Linux内核同步互斥机制:自旋锁、互斥锁、信号量、原子操作和completion
同步互斥
Linux内核里有多个"执行路径"可能同时访问同一个数据:
- 多CPU并发:CPU0在跑一个函数,CPU1同时在跑一个中断处理函数,两个都在操作UCR1寄存器
- 进程与中断并发:代码正在执行,硬件中断来了,CPU暂停当前代码去执行中断处理函数,中断处理函数里也访问同样的数据
- 进程与进程并发:两个用户程序同时操作同一个串口设备
当两个执行路径同时读-改-写同一个变量或寄存器时,就会出现竞态条件。经典例子:
假设count当前值是5
路径A:读count(得到5)→ 加1 → 写回count(写入6)
路径B:读count(得到5)→ 加1 → 写回count(写入6)
结果:两个路径各加了1,但count只变成了6而不是7
同步互斥机制就是确保"读-改-写"这种操作不被打断,同一时刻只有一个路径在操作共享资源。
一、自旋锁(spinlock)
基本原理
自旋锁是一个锁变量。当路径A获取(lock)了这个锁,路径B也想获取时,B不会睡眠,而是在原地不停循环检查(自旋/spinning)锁是否被释放了。一旦A释放(unlock)了锁,B立刻获取到,继续执行。
定义和初始化
静态定义(全局变量或结构体成员,编译时就定好):
/* 方式一:定义并初始化为一体 */
static DEFINE_SPINLOCK(my_lock);
/* 方式二:先定义,再初始化(用于结构体成员) */
struct my_device {
spinlock_t lock;
int data;
};
/* 在probe或init函数里初始化 */
spin_lock_init(&dev->lock);
使用方式(4个变体)
变体1:spin_lock / spin_unlock(最基本)
spin_lock(&my_lock);
/* 临界区:操作共享资源 */
spin_unlock(&my_lock);
只保护多CPU之间的并发。如果路径A持锁时被本CPU的中断打断,中断处理函数也要获取同一个锁——死锁。
适用场景:共享资源只在进程上下文之间竞争,不涉及中断。实际驱动开发中很少单独用这个。
变体2:spin_lock_irqsave / spin_unlock_irqrestore(最常用)
unsigned long flags;
spin_lock_irqsave(&my_lock, flags);
/* 临界区:操作共享资源 */
spin_unlock_irqrestore(&my_lock, flags);
获取锁的同时关闭本CPU中断并保存中断状态到flags,释放锁时恢复中断状态。
为什么要保存/恢复而不是简单开关?举个例子:
void func_a(void) {
spin_lock_irqsave(&lock, flags); /* 假设进来之前中断是开的,flags记录"开" */
func_b();
spin_unlock_irqrestore(&lock, flags); /* 恢复成"开" */
}
void func_b(void) {
spin_lock_irqsave(&lock2, flags2); /* 进来之前中断已经被func_a关了,flags2记录"关" */
/* ... */
spin_unlock_irqrestore(&lock2, flags2); /* 恢复成"关",不会意外开中断 */
}
如果func_b里用的是spin_lock_irq/spin_unlock_irq(无条件关/开中断),那func_b结束时会无条件开中断,但func_a还没释放锁呢,中断就进来了,可能死锁。所以irqsave/irqrestore是最安全的。
适用场景:共享资源在进程上下文和中断上下文之间竞争。你的UART驱动就是这个场景。这是驱动开发中最常用的自旋锁变体。
变体3:spin_lock_irq / spin_unlock_irq
spin_lock_irq(&my_lock); /* 获取锁 + 无条件关中断 */
/* 临界区 */
spin_unlock_irq(&my_lock); /* 释放锁 + 无条件开中断 */
跟irqsave的区别:不保存中断状态,释放时无条件开中断。只有确定调用这段代码之前中断一定是开的时候才能用。在不确定的情况下用irqsave更安全。
变体4:spin_lock_bh / spin_unlock_bh
spin_lock_bh(&my_lock); /* 获取锁 + 关软中断(bottom half) */
/* 临界区 */
spin_unlock_bh(&my_lock); /* 释放锁 + 开软中断 */
bh是bottom half(软中断/tasklet)的意思。只关闭软中断,不关硬中断。
适用场景:共享资源在进程上下文和软中断之间竞争(比如网络驱动里经常用)。
自旋锁的使用规则
- 持锁期间绝对不能睡眠。不能调用任何可能睡眠的函数:kmalloc(GFP_KERNEL)、msleep、copy_from_user、mutex_lock等等。因为睡眠意味着CPU切换去跑其他进程,而其他进程可能也要获取这个锁→死锁
- 临界区要尽可能短。因为其他路径在busy-wait空转等,浪费CPU时间
- 如果共享资源涉及中断上下文,必须用irqsave变体。否则会死锁
- 同一个锁不能嵌套获取。路径A持有锁L,再次spin_lock(&L)→死锁
二、互斥锁(mutex)
基本原理
互斥锁跟自旋锁的目的一样——保证同一时刻只有一个路径操作共享资源。区别是等待方式:拿不到锁时不是忙等,而是睡眠。当前进程被放进等待队列,CPU去执行其他进程。锁释放后,等待的进程被唤醒。
定义和初始化
/* 方式一:静态定义并初始化 */
static DEFINE_MUTEX(my_mutex);
/* 方式二:动态初始化(结构体成员) */
struct my_device {
struct mutex lock;
char *buffer;
};
mutex_init(&dev->lock);
使用方式
基本用法:mutex_lock / mutex_unlock
mutex_lock(&my_mutex);
/* 临界区:可以做耗时操作,可以睡眠 */
mutex_unlock(&my_mutex);
如果锁已被持有,mutex_lock会让当前进程睡眠等待,直到锁被释放。
可中断版本:mutex_lock_interruptible
int ret = mutex_lock_interruptible(&my_mutex);
if (ret) {
/* 被信号打断了,没拿到锁,返回-ERESTARTSYS */
return ret;
}
/* 拿到了锁 */
mutex_unlock(&my_mutex);
普通mutex_lock在等待时,即使用户按Ctrl+C也不会响应。mutex_lock_interruptible可以被信号打断,更适合用户态可能长时间等待的场景。
尝试获取:mutex_trylock
if (mutex_trylock(&my_mutex)) {
/* 拿到了锁 */
mutex_unlock(&my_mutex);
} else {
/* 没拿到,但不等待,立刻返回做其他事 */
}
互斥锁的使用规则
- 只能在进程上下文使用,绝不能在中断上下文使用。因为中断上下文不能睡眠,而mutex等待时会睡眠
- 持锁进程可以睡眠(跟spinlock的最大区别)
- 同一个mutex不能嵌套获取。持有者再次mutex_lock→死锁
- 谁获取谁释放。路径A拿的mutex只能路径A来unlock,不能让路径B代替unlock
什么场景用mutex
典型场景:一个字符设备驱动的read/write函数里保护buffer。因为read/write是在进程上下文执行的,里面可能要copy_to_user/copy_from_user(可能睡眠),这种情况spinlock不行(不能睡眠),只能用mutex。
static ssize_t my_read(struct file *filp, char __user *buf, size_t count, loff_t *ppos)
{
struct my_device *dev = filp->private_data;
mutex_lock(&dev->lock);
/* copy_to_user可能睡眠(比如用户空间页面被换出,需要page fault) */
if (copy_to_user(buf, dev->buffer, count)) {
mutex_unlock(&dev->lock);
return -EFAULT;
}
mutex_unlock(&dev->lock);
return count;
}
三、信号量(semaphore)
基本原理
信号量内部维护一个计数器。每次获取(down)计数器减1,每次释放(up)计数器加1。当计数器减到0时,后来的获取者睡眠等待(这点跟mutex一样,不是忙等)。
mutex本质上是计数器初始值为1的信号量——同一时刻只有1个路径能持有。而信号量可以设初始值为N,允许N个路径同时访问。
定义和初始化
#include <linux/semaphore.h>
/* 方式一:静态定义,计数初始值为N */
static DEFINE_SEMAPHORE(my_sem); /* 默认初始值为1,等同于mutex */
/* 方式二:动态初始化,指定初始计数值 */
struct semaphore my_sem;
sema_init(&my_sem, 3); /* 允许最多3个路径同时持有 */
使用方式
/* 获取信号量(计数器减1,如果已经是0就睡眠等待) */
down(&my_sem);
/* 临界区 */
up(&my_sem); /* 释放信号量(计数器加1,唤醒等待者) */
/* 可中断版本 */
if (down_interruptible(&my_sem)) {
/* 被信号打断了,没拿到信号量 */
return -ERESTARTSYS;
}
/* 临界区 */
up(&my_sem);
/* 尝试获取(不等待,立即返回) */
if (down_trylock(&my_sem)) {
/* 没拿到 */
} else {
/* 拿到了 */
up(&my_sem);
}
信号量的使用规则
- 跟mutex一样不能在中断上下文使用(等待时会睡眠)
- up和down不要求是同一个路径(这点跟mutex不同——mutex要求谁lock谁unlock,信号量可以一个路径down另一个路径up)
- 可以设初始值大于1,允许多个并发访问
信号量的典型场景
场景1:限制并发访问数量。 比如一个设备最多支持3个进程同时打开:
static DEFINE_SEMAPHORE(dev_sem); /* 假设sema_init为3 */
static int my_open(struct inode *inode, struct file *filp)
{
if (down_trylock(&dev_sem))
return -EBUSY; /* 已经有3个进程在用了 */
return 0;
}
static int my_release(struct inode *inode, struct file *filp)
{
up(&dev_sem);
return 0;
}
场景2:生产者-消费者同步。 一个路径产生数据后up,另一个路径down等待数据到来。这里up和down不是同一个路径,所以不能用mutex,只能用信号量。
四、原子操作(atomic)
基本原理
原子操作是把一个简单操作(加1、减1、读、写、测试并设置等)变成CPU硬件级别的不可分割的单条指令。不需要锁,没有获取/释放的开销。
前面说的count竞态问题——"读count、加1、写回count"是三步操作可能被打断。原子操作让"读-改-写"变成一条指令,硬件保证不会被打断。
原子整数操作
#include <linux/atomic.h>
/* 定义和初始化 */
atomic_t counter = ATOMIC_INIT(0); /* 初始值为0 */
/* 基本操作 */
atomic_set(&counter, 5); /* 设置为5 */
int val = atomic_read(&counter); /* 读取当前值,返回5 */
atomic_inc(&counter); /* 加1,变成6 */
atomic_dec(&counter); /* 减1,变回5 */
atomic_add(3, &counter); /* 加3,变成8 */
atomic_sub(2, &counter); /* 减2,变成6 */
/* 操作并返回结果 */
int new_val = atomic_inc_return(&counter); /* 加1并返回新值 */
/* 测试操作 */
if (atomic_dec_and_test(&counter)) {
/* 减1后值变成0了,返回true */
/* 典型用途:引用计数减到0时释放资源 */
}
原子位操作
除了整数,内核也提供对单个bit的原子操作:
#include <linux/bitops.h>
unsigned long flags = 0;
set_bit(3, &flags); /* 把第3位设为1 */
clear_bit(3, &flags); /* 把第3位设为0 */
change_bit(3, &flags); /* 翻转第3位 */
if (test_bit(3, &flags)) {
/* 第3位是1 */
}
/* 测试并设置(原子地检查旧值并设新值) */
if (test_and_set_bit(3, &flags)) {
/* 之前第3位就是1(已经被别人设过了) */
} else {
/* 之前第3位是0,现在被我设成1了 */
}
原子操作的适用场景
场景1:引用计数。 比如一个设备被多少个进程打开了:
static atomic_t open_count = ATOMIC_INIT(0);
static int my_open(struct inode *inode, struct file *filp)
{
atomic_inc(&open_count);
pr_info("device opened %d times\n", atomic_read(&open_count));
return 0;
}
static int my_release(struct inode *inode, struct file *filp)
{
atomic_dec(&open_count);
return 0;
}
场景2:简单标志位。 比如标记设备是否正在传输中:
static atomic_t busy = ATOMIC_INIT(0);
if (atomic_cmpxchg(&busy, 0, 1) == 0) {
/* 成功从0改成1,说明之前没人在用,我拿到了 */
do_transfer();
atomic_set(&busy, 0);
} else {
/* 之前已经是1了,有人在用 */
return -EBUSY;
}
原子操作的局限
原子操作只能保护单个变量的单次操作。如果你需要保护"读寄存器A → 根据结果修改寄存器B"这种多步操作,原子操作无能为力,必须用spinlock或mutex。
五、其他机制
读写自旋锁(rwlock / rwsem)
读写锁区分"读者"和"写者"。多个读者可以同时持有锁(因为只读不冲突),但写者必须独占。
/* 读写自旋锁 */
rwlock_t my_rwlock;
rwlock_init(&my_rwlock);
read_lock(&my_rwlock); /* 读者获取:多个读者可以同时持有 */
/* 读共享数据 */
read_unlock(&my_rwlock);
write_lock(&my_rwlock); /* 写者获取:必须等所有读者和写者都释放 */
/* 修改共享数据 */
write_unlock(&my_rwlock);
/* 也有sleepable版本:读写信号量 rw_semaphore */
struct rw_semaphore my_rwsem;
init_rwsem(&my_rwsem);
down_read(&my_rwsem); /* 读者 */
up_read(&my_rwsem);
down_write(&my_rwsem); /* 写者 */
up_write(&my_rwsem);
适用场景:读多写少的数据结构,比如系统的路由表、进程列表。读者之间不互斥可以提高并发性能。同样有read_lock_irqsave等变体。
completion(完成量)
一个路径等待另一个路径通知"事情做完了"。
#include <linux/completion.h>
struct completion my_done;
init_completion(&my_done);
/* 路径A:等待完成(会睡眠) */
wait_for_completion(&my_done);
/* 或者带超时版本 */
unsigned long timeout = wait_for_completion_timeout(&my_done, msecs_to_jiffies(1000));
if (timeout == 0) {
/* 超时了,1秒内没等到 */
}
/* 路径B(比如中断处理函数):通知完成 */
complete(&my_done);
跟信号量的区别:completion专门为"等待-通知"场景设计,语义更清晰,内核内部实现也做了优化。信号量虽然也能实现类似功能(down等待,up通知),但completion是更推荐的做法。
RCU(Read-Copy-Update)
RCU是一种极端优化读性能的机制:读者几乎零开销(不需要任何锁操作),写者负责更复杂的更新流程(先复制一份数据,修改副本,然后替换指针,等所有读者都不再引用旧数据后释放旧数据)。
六、死锁(deadlock)
死锁就是两个或多个执行路径互相等待对方释放资源,谁也无法继续。
场景1:自死锁(同一个锁获取两次)
spin_lock(&lock_A);
/* 做一些操作... */
spin_lock(&lock_A); /* 自己已经持有lock_A,再获取→永远等不到→死锁 */
怎么发生的:通常不是这么直白写的,而是函数调用链里隐藏的。比如func_a持有lock_A,func_a调用func_b,func_b里面也spin_lock(&lock_A)。写代码时可能没注意到func_b里有这个锁。
预防:用锁之前明确哪些函数会持有哪个锁,不要在持锁路径上调用可能获取同一个锁的函数。
场景2:ABBA死锁(两个锁交叉获取)
/* CPU 0 */ /* CPU 1 */
spin_lock(&lock_A); spin_lock(&lock_B);
/* ... */ /* ... */
spin_lock(&lock_B); /* 等CPU1释放B */ spin_lock(&lock_A); /* 等CPU0释放A */
/* 互相等,死锁 */
预防:规定全局的锁获取顺序。如果所有代码路径都先获取A再获取B,就不会出现交叉。
场景3:进程上下文和中断上下文死锁
/* 进程上下文 */
spin_lock(&lock); /* 获取了锁 */
/* 这时候中断来了,CPU去执行中断处理函数 */
/* 中断处理函数 */
spin_lock(&lock); /* 等待锁释放——但持锁的进程被打断了,没法释放→死锁 */
预防:如果一个锁会被中断处理函数使用,进程上下文里必须用spin_lock_irqsave(获取锁的同时关中断),这样中断就不会在持锁期间进来。
场景4:持spinlock时睡眠
spin_lock(&lock);
kmalloc(size, GFP_KERNEL); /* GFP_KERNEL可能导致当前进程睡眠 */
/* 进程睡了,锁还没释放。其他路径等这个锁就要一直忙等 */
/* 如果调度到的其他进程也要这个锁→死锁或长时间卡住 */
spin_unlock(&lock);
预防:持spinlock期间只调用不会睡眠的函数。需要分配内存就用GFP_ATOMIC(不睡眠版本)。
openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构
更多推荐

所有评论(0)