同步互斥

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)的意思。只关闭软中断,不关硬中断。

适用场景:共享资源在进程上下文和软中断之间竞争(比如网络驱动里经常用)。

自旋锁的使用规则

  1. 持锁期间绝对不能睡眠。不能调用任何可能睡眠的函数:kmalloc(GFP_KERNEL)、msleep、copy_from_user、mutex_lock等等。因为睡眠意味着CPU切换去跑其他进程,而其他进程可能也要获取这个锁→死锁
  2. 临界区要尽可能短。因为其他路径在busy-wait空转等,浪费CPU时间
  3. 如果共享资源涉及中断上下文,必须用irqsave变体。否则会死锁
  4. 同一个锁不能嵌套获取。路径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 {
    /* 没拿到,但不等待,立刻返回做其他事 */
}

互斥锁的使用规则

  1. 只能在进程上下文使用,绝不能在中断上下文使用。因为中断上下文不能睡眠,而mutex等待时会睡眠
  2. 持锁进程可以睡眠(跟spinlock的最大区别)
  3. 同一个mutex不能嵌套获取。持有者再次mutex_lock→死锁
  4. 谁获取谁释放。路径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);
}

信号量的使用规则

  1. 跟mutex一样不能在中断上下文使用(等待时会睡眠)
  2. up和down不要求是同一个路径(这点跟mutex不同——mutex要求谁lock谁unlock,信号量可以一个路径down另一个路径up)
  3. 可以设初始值大于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(不睡眠版本)。

Logo

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐