Linux并发与竞争

---

时间轴

时间轴

2025-11-12

init

2026-05-21

add rwlock, rw_semaphore , rcu and per_cpu

---

Linux 驱动笔记

目录	链接
1. Linux 驱动框架	Linux 驱动框架 https://even629.com/posts/2511093/
2. Linux 驱动加载逻辑	Linux 驱动加载逻辑 https://even629.com/posts/2511100/
3. 字符设备基础	字符设备基础 https://even629.com/posts/2511113/
4. 并发与竞争	并发与竞争 https://even629.com/posts/2511123/
5. 高级字符设备进阶	高级字符设备进阶 https://even629.com/posts/2511133/
6. 中断	中断 https://even629.com/posts/2511143/
7. 平台总线	平台总线 https://even629.com/posts/2511153/
8. 设备树	设备树 https://even629.com/posts/2511300/
9. 设备模型	设备模型 https://even629.com/posts/2512013/
10. 热插拔	热插拔 https://even629.com/posts/2512023/
11. pinctrl 子系统	pinctrl 子系统 https://even629.com/posts/2512160/
12. gpio 子系统	gpio 子系统 https://even629.com/posts/2512173/
13. 输入子系统	输入子系统 https://even629.com/posts/2512183/
14. 单总线	单总线 https://even629.com/posts/2512233/
15. I2C	I2C https://even629.com/posts/2512283/
16. SPI	SPI https://even629.com/posts/2512303/
17. UART	UART https://even629.com/posts/2512313/
18. PWM	PWM https://even629.com/posts/2601043/
19. RTC	RTC https://even629.com/posts/2601053/
20. Watchdog	Watchdog https://even629.com/posts/2601063/
21. CAN	CAN https://even629.com/posts/2601093/
22. 网络设备	网络设备 https://even629.com/posts/2601133/
23. ADC	ADC https://even629.com/posts/2601143/
24. IIO	IIO https://even629.com/posts/2602153
25. USB	USB https://even629.com/posts/2602163
26. LCD	LCD https://even629.com/posts/2602173

---

并发与竞争概念

下面把并发与并行统称并发。

在并发执行的环境中，多个程序可能同时访问同一个共享资源。当多个任务试图同时操作这类资源时，可能会出现执行异常或数据错误，这类问题被称为竞争。

竞争产生的常见原因包括：

1. 多线程访问。由于 Linux 是一个多任务操作系统，多个线程可能同时访问同一共享资源，这是竞争产生的基本原因。
2. 中断访问。当进程正在访问某一共享资源，而中断打断了正在执行的进程，而发出中断的进程与被打断的进程之间也可能发生竞争。
3. 抢占访问。linux2.6 及更高版本引入了抢占式内核，高优先级的任务可以打断低优先级的任务。如果正在访问共享资源的任务被打断，另一个任务接着访问相同资源，就可能引发竞争。
4. 多核并发访问（SMP）。在多核处理器系统中，不同 CPU 核心之间可能并发访问同一共享资源，形成核间竞争。

Linux 内核提供了多种机制来应对这一问题，常用的方法包括原子操作、自旋锁、互斥锁、信号量等。

原子操作

原子操作会将对整形变量的一次读写操作视为一个整体，确保其不可分割，从而避免竞争。

原子操作可以进一步细分为“整型原子操作”和“位原子操作”，这里首先对整型原子操作进行讲解。

整形原子操作

在 Linux 内核中使用 atomic_t 和 atomic64_t 结构体分别来定义 32 位系统和 64 位系统下的原子变量

include/linux/types.h

typedef struct {
	int counter;
} atomic_t;

#define ATOMIC_INIT(i) { (i) }

#ifdef CONFIG_64BIT
typedef struct {
	s64 counter;
} atomic64_t;
#endif

32 位的 atomic_t

include/linux/atomic.h

函数原型	功能解释	参数	返回值
#define ATOMIC_INIT(i)	初始化 atomic_t 变量为 i	i：初始整数值	无
int atomic_read(const atomic_t *v)	读取原子变量的值	v：atomic_t 指针	返回 v 的当前值
void atomic_set(atomic_t *v, int i)	设置原子变量为 i	v：变量指针；i：要写入的值	无
void atomic_add(int i, atomic_t *v)	原子地将 v += i	i：增加值；v：变量指针	无
void atomic_sub(int i, atomic_t *v)	原子地将 v -= i	i：减少值；v：变量指针	无
void atomic_inc(atomic_t *v)	原子地将 v++	v：变量指针	无
void atomic_dec(atomic_t *v)	原子地将 v--	v：变量指针	无
int atomic_inc_return(atomic_t *v)	原子 v++ 并返回新值（先加后返回）	v：变量指针	自增后的值
int atomic_dec_return(atomic_t *v)	原子 v-- 并返回新值（先减后返回）	v：变量指针	自减后的值
int atomic_sub_and_test(int i, atomic_t *v)	原子 v -= i，若结果为 0 返回 true	i：减的值；v：变量指针	为 0 → 返回 1，不为 0 → 返回 0
int atomic_dec_and_test(atomic_t *v)	原子 v--，若结果为 0 返回 true	v：变量指针	为 0 → 返回 1，不为 0 → 返回 0
int atomic_inc_and_test(atomic_t *v)	原子 v++，若结果为 0 返回 true	v：变量指针	为 0 → 返回 1，否则 0
int atomic_add_negative(int i, atomic_t *v)	原子 v += i，若结果 < 0 返回 true	i：增加值；v：变量指针	结果 < 0 → 返回 1，否则 0

64 位的 atomic64_t

函数原型	功能解释	参数	返回值
#define ATOMIC64_INIT(i)	初始化 atomic64_t 为 i	i：初始 64 位整数	无
long long atomic64_read(const atomic64_t *v)	读取 64 位原子变量的值	v：变量指针	返回当前 64 位值
void atomic64_set(atomic64_t *v, long long i)	设置原子变量为 i	v：变量指针；i：写入的 64 位值	无
void atomic64_add(long long i, atomic64_t *v)	原子执行 v += i	i：增加值；v：变量指针	无
void atomic64_sub(long long i, atomic64_t *v)	原子执行 v -= i	i：减少值；v：变量指针	无
void atomic64_inc(atomic64_t *v)	原子执行 v++	v：变量指针	无
void atomic64_dec(atomic64_t *v)	原子执行 v--	v：变量指针	无
long long atomic64_add_return(long long i, atomic64_t *v)	原子执行 v += i 并返回新值（先加后返回）	i、v	返回加后的值
long long atomic64_sub_return(long long i, atomic64_t *v)	原子执行 v -= i 并返回新值（先减后返回）	i、v	返回减后的值
long long atomic64_inc_return(atomic64_t *v)	原子 v++ 并返回新值	v	返回自增后的值
long long atomic64_dec_return(atomic64_t *v)	原子 v-- 并返回新值	v	返回自减后的值
int atomic64_add_negative(long long i, atomic64_t *v)	原子 v += i，若结果 < 0 返回 true	i、v	结果 < 0 → 1，否则 0
int atomic64_add_and_test(long long i, atomic64_t *v)	原子 v += i，若结果 == 0 返回 true	i、v	结果 == 0 → 1，否则 0
int atomic64_sub_and_test(long long i, atomic64_t *v)	原子 v -= i，若结果 == 0 返回 true	i、v	结果 == 0 → 1，否则 0
int atomic64_inc_and_test(atomic64_t *v)	原子 v++，若结果 == 0 返回 true	v	为 0 → 1，否则 0
int atomic64_dec_and_test(atomic64_t *v)	原子 v--，若结果 == 0 返回 true	v	为 0 → 1，否则 0
long long atomic64_xchg(atomic64_t *v, long long i)	原子交换：将 v 设置成 i 并返回旧值	v、i	返回交换前的旧值
long long atomic64_cmpxchg(atomic64_t *v, long long old, long long new)	原子比较交换：若 *v == old，则写入 new	v、old、new	返回交换前的旧值（可用于判断是否成功）

错误用法

下面的使用方法是错误的：

static atomic_t atomic_key = ATOMIC_INIT(1);

int atomic_t_test_open(struct inode *inode, struct file *file)
{
        file->private_data = test_drv_data;
        if(atomic_read(&atomic_key) == 0){
                return -EBUSY;
        }
        atomic_dec(&atomic_key);

        return 0;
}

int atomic_t_test_release(struct inode *inode, struct file *file)
{
        atomic_inc(&atomic_key);
        return 0;
}

原因是atomic_read(&atomic_key)和atomic_dec(&atomic_key)这两个操作是分开的，两个进程可以同时atomic_read(&atomic_key)然后atomic_dec(&atomic_key)成功。可以把这两个操作合到一起

if (!atomic_dec_and_test(&atomic_key))
    return -EBUSY;

例子：

#include <linux/init.h>
#include <linux/module.h>
#include <linux/cdev.h>
#include <linux/slab.h>
#include <linux/atomic.h>

struct test_drv_data {
        dev_t dev_num;
        struct cdev cdev;
        struct class *class;
        struct device *dev;
};

static struct test_drv_data *test_drv_data;

static atomic_t atomic_key = ATOMIC_INIT(1);

int atomic_t_test_open(struct inode *inode, struct file *file)
{
        file->private_data = test_drv_data;
        if (atomic_dec_and_test(&atomic_key) == 0){
                pr_err("this device is opened by another process\n");
                return -EBUSY;
        }
        pr_info("atomic_t_test_open is called\n");
        return 0;
}

int atomic_t_test_release(struct inode *inode, struct file *file)
{
        atomic_inc(&atomic_key);
        return 0;
}

ssize_t atomic_t_test_read (struct file *file, char __user *buf, size_t size, loff_t *offset){
	pr_info("atomic_t_test_read is called\n");
	return 0;
}
ssize_t atomic_t_test_write (struct file *file, const char __user *buf, size_t size, loff_t *offset){
	pr_info("atomic_t_test_write is called\n");
	return 0;
}

struct file_operations fops = {
        .owner = THIS_MODULE,
        .open = atomic_t_test_open,
        .release = atomic_t_test_release,
};

static int __init atomic_t_test_init(void)
{
        int err;
        test_drv_data = (struct test_drv_data *)kzalloc(sizeof(struct test_drv_data), GFP_KERNEL);
        if (test_drv_data == NULL) {
                err = -ENOMEM;
                goto kzalloc_fail;
        }
        err = alloc_chrdev_region(&test_drv_data->dev_num, 0, 1, "atomic_t test chrdev region");
        if (err < 0)
                goto alloc_chrdev_region_fail;
        cdev_init(&test_drv_data->cdev, &fops);
        err = cdev_add(&test_drv_data->cdev, test_drv_data->dev_num, 1);
        if (err < 0)
                goto cdev_add_fail;
        test_drv_data->class = class_create(THIS_MODULE, "atomic_t_test");
        if (IS_ERR(test_drv_data->class)) {
                err = PTR_ERR(test_drv_data->class);
                goto class_create_fail;
        }
        test_drv_data->dev = device_create(test_drv_data->class, NULL, test_drv_data->dev_num, NULL,
                                           "atomic_t_test%d", 0);
        if (IS_ERR(test_drv_data->dev)) {
                err = PTR_ERR(test_drv_data->dev);
                goto device_create_fail;
        }

        return 0;
device_create_fail:
        class_destroy(test_drv_data->class);
class_create_fail:
        cdev_del(&test_drv_data->cdev);
cdev_add_fail:
        unregister_chrdev_region(test_drv_data->dev_num, 1);
alloc_chrdev_region_fail:
        kfree(test_drv_data);
kzalloc_fail:
        return err;
}

static void __exit atomic_t_test_exit(void)
{
        device_destroy(test_drv_data->class, test_drv_data->dev_num);
        class_destroy(test_drv_data->class);
        cdev_del(&test_drv_data->cdev);
        unregister_chrdev_region(test_drv_data->dev_num, 1);
        kfree(test_drv_data);
}

module_init(atomic_t_test_init);
module_exit(atomic_t_test_exit);

MODULE_LICENSE("GPL");
MODULE_AUTHOR("even629<asqwgo@163.com>");
MODULE_DESCRIPTION("This is a test sample for atomic_t");

测试代码：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>

int main(int argc, char **argv)
{
        pid_t pid;
        int fd;

        pid = fork();

        if (pid < 0) {
                goto fail;
        } else if (pid == 0) { // child process
                fd = open("/dev/atomic_t_test0", O_RDWR);
                if (fd < 0)
                        goto fail;

                printf("child open device success\n");
                // do something
                sleep(3);
                close(fd);

        } else { // parent process
                fd = open("/dev/atomic_t_test0", O_RDWR);
                if(fd < 0){
                        goto fail;
                }
                printf("parent open device success\n");
                // do something
                sleep(3);
                close(fd);
        }

        return 0;
fail:
        fprintf(stderr, "Error:%s[errno:%d]\n", strerror(errno), errno);
        exit(EXIT_FAILURE);
}

测试：

$ insmod atomic_t.ko
[   13.631548] atomic_t: loading out-of-tree module taints kernel.
$ ./test_atomic_t.o
[   20.232024] atomic_t_test_open is called
[   20.233129] this device is opened by another process
parent open device success
Error:Device or resource busy[errno:16]

atomic_cmpxchg

这个函数比较难理解，作用是：

• 如果原子变量的当前值等于 old
  • 把它改成 new
  • 返回“修改前的值”(因为v修改前的值等于old，所以返回的值等于old)
• 如果当前值 ≠ old：
  • 不会修改原子变量
  • 直接返回“当前值”

#include <linux/atomic.h>

long long atomic64_cmpxchg(atomic64_t *v, long long old, long long new);

参数

参数	说明
v	指向要操作的 atomic64_t 变量的指针
old	期望的旧值（expected value）
new	如果原子变量的值等于 old，就用 new 替换它

返回值

• 返回 原子变量操作前的原始值。
• 如果原始值等于 old，把原始值替换位 new，返回值 = old。
• 如果原始值不等于 old，替换失败，返回值 ≠ old。

例子

#include <linux/init.h>
#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/moduleparam.h>
#include <linux/fs.h>
#include <linux/types.h>
#include <linux/cdev.h>
#include <linux/device.h>
#include <linux/uaccess.h>
#include <linux/err.h>
#include <linux/errno.h>
#include <linux/atomic.h>

struct device_test {
	// 设备号
	dev_t dev_id;
	// 所属类
	struct class *class;
	// 类下的device 设备节点
	struct device *device;
	// cdev,字符类设备
	struct cdev cdev_test;
};

static struct device_test device1;

static atomic64_t cnt = ATOMIC_INIT(1);

static int major_num;
static int minor_num;

module_param(major_num, int, S_IRUGO);
module_param(minor_num, int, S_IRUGO);

int cdev_test_open(struct inode *inode, struct file *file)
{
	if (atomic64_cmpxchg(&cnt, 1, 0) != 1) {
		return -EBUSY;
	}
	file->private_data = &device1;
	pr_info("cdev_test_open was called");
	return 0;
}

ssize_t cdev_test_read(struct file *file, char __user *buf, size_t size,
		       loff_t *off)
{
	//struct device_test *dev1 = (struct device_test *)file->private_data;

	pr_info("cdev_test_read was called");
	return 0;
}

ssize_t cdev_test_write(struct file *file, const char __user *buf, size_t size,
			loff_t *off)
{
	//struct device_test *dev1 = (struct device_test *)file->private_data;

	return 0;
}
int cdev_test_release(struct inode *inode, struct file *file)
{
	atomic64_set(&cnt, 1);
	pr_info("cdev_test_release was called");
	return 0;
}

static struct file_operations cdev_test_ops = { .owner = THIS_MODULE,
						.open = cdev_test_open,
						.read = cdev_test_read,
						.write = cdev_test_write,
						.release = cdev_test_release };

static int __init my_driver_init(void)
{
	int ret;

	// 申请设备号
	if (major_num) { //驱动传参，静态申请设备号
		device1.dev_id = MKDEV(major_num, minor_num);

		pr_info("major from module_param: %d", MAJOR(device1.dev_id));
		pr_info("minor from module_param: %d", MINOR(device1.dev_id));

		ret = register_chrdev_region(device1.dev_id, 1,
					     "my_driver device");

		if (ret < 0) {
			pr_err("register_chrdev_region error\n");
			goto get_chrdev_region_err;
		}
		pr_info("register_chrdev_region ok\n");

	} else { //动态申请设备号
		ret = alloc_chrdev_region(&device1.dev_id, 0, 1,
					  "my_driver device");
		if (ret < 0) {
			pr_err("alloc_chrdev_region error\n");
			goto get_chrdev_region_err;
		}
		pr_info("alloc_chrdev_region ok\n");

		pr_info("major allocated: %d", MAJOR(device1.dev_id));
		pr_info("minor allocated: %d", MINOR(device1.dev_id));
	}
	// cdev初始化，注册字符类设备
	cdev_init(&device1.cdev_test, &cdev_test_ops);

	//将 owner 字段指向本模块，可以避免在模块的操作正在被使用时卸载该模块
	device1.cdev_test.owner = THIS_MODULE;
	// 添加一个字符设备
	ret = cdev_add(&device1.cdev_test, device1.dev_id, 1);

	if (ret < 0) {
		pr_err("cdev_add error\n");
		goto cdev_add_err;
	}
	pr_info("cdev_add ok\n");

	// 创建一个class
	device1.class = class_create(THIS_MODULE, "test");
	if (IS_ERR(device1.class)) {
		ret = PTR_ERR(device1.class);
		pr_err("class create error\n");
		goto class_create_err;
	}
	// 在class下创建一个设备结点, /dev/my_driver
	device1.device = device_create(device1.class, NULL, device1.dev_id,
				       NULL, "my_driver");
	if (IS_ERR(device1.device)) {
		ret = PTR_ERR(device1.device);
		pr_err("device create error\n");
		goto device_create_err;
	}
	pr_info("my_driver: Module loaded\n");

	return 0;

device_create_err:
	class_destroy(device1.class);
class_create_err:
	cdev_del(&device1.cdev_test);
cdev_add_err:
	unregister_chrdev_region(device1.dev_id, 1);
get_chrdev_region_err:
	return ret;
}

static void __exit my_driver_exit(void)
{
	// 删除设备结点
	device_destroy(device1.class, device1.dev_id);
	// 删除该设备结点的class
	class_destroy(device1.class);
	// 删除cdev
	cdev_del(&device1.cdev_test);
	// 释放设备号
	unregister_chrdev_region(device1.dev_id, 1);

	pr_info("unregister_chrdev_region ok\n");
	pr_info("my_driver: Module unloaded\n");
}

module_init(my_driver_init);
module_exit(my_driver_exit);

MODULE_LICENSE("GPL");
MODULE_AUTHOR("Zhao Hang");
MODULE_DESCRIPTION("my_driver Kernel Module");

位原子操作

函数原型	功能解释	参数说明	返回值
void set_bit(int nr, void *p)	将 p 地址的第 nr 位 置为 1	nr：位编号（0 表最低位）p：起始地址	无
void clear_bit(int nr, void *p)	将 p 地址的第 nr 位 清零	同上	无
void change_bit(int nr, void *p)	将 nr 位 取反（0→1 或 1→0）	同上	无
int test_bit(int nr, void *p)	读取 nr 位的值	同上	返回该位的值：0 或 1
int test_and_set_bit(int nr, void *p)	将 nr 位设置为 1，并返回原值	同上	返回操作前该位的值（0 或 1）
int test_and_clear_bit(int nr, void *p)	将 nr 位清零，并返回原值	同上	返回操作前该位的值
int test_and_change_bit(int nr, void *p)	将 nr 位翻转（1↔0），并返回原值	同上	返回操作前该位的值

位编号是从 p 指向内存的起始地址算起：

• nr = 0 → 第一字节的最低位（bit0）
• nr = 7 → 第一字节的最高位
• nr = 8 → 第二字节的 bit0
• 依此类推

位编号为绝对 bit index，而不是字节内偏移。

自旋锁

自旋锁是为了保护共享资源提出的一种锁机制。当一个线程尝试获取自旋锁而发现该锁已被其他线程持有时，它不会进入睡眠状态等待，而是会持续循环地尝试获取锁，直到成功为止。这个过程称为“自旋”，因此得名“自旋锁”。

定义

include/linux/spinlock_types.h

typedef struct spinlock {
	union {
		struct raw_spinlock rlock;

#ifdef CONFIG_DEBUG_LOCK_ALLOC
# define LOCK_PADSIZE (offsetof(struct raw_spinlock, dep_map))
		struct {
			u8 __padding[LOCK_PADSIZE];
			struct lockdep_map dep_map;
		};
#endif
	};
} spinlock_t;

offsetof 定义在include/linux/stddef.h。作用是计算结构体到某一个字段的偏移。

#ifdef __compiler_offsetof
#define offsetof(TYPE, MEMBER)	__compiler_offsetof(TYPE, MEMBER)
#else
#define offsetof(TYPE, MEMBER)	((size_t)&((TYPE *)0)->MEMBER)
#endif

上述 spinlock 字段通过 offsetof 计算设置填充字段，保证两个 union 成员的 dep_map 地址完全一致，这样保证不管是否启用 CONFIG_DEBUG_LOCK_ALLOC ，spinlock_t 的大小和对齐方式 一致

还有另外一个作用：

spinlock_t 是 通用锁类型

• 不想暴露 raw_spinlock 的内部结构
• 但 lockdep 需要 dep_map

这样设计让内核代码可以这样写：

lockdep_init_map(&lock->dep_map, ...);

而非：

lock->rlock.dep_map   // ❌ 不希望这样

自旋锁相关 API 函数定义在内核源码include/linux/spinlock.h 文件中（spinlock.h 头文件包含 spinlock_types.h 等，所以只需加入 spinlock.h 头文件即可）

API

API / 宏	返回值类型	功能描述	是否阻塞	中断状态	返回值 / 备注
DEFINE_SPINLOCK(name)	void	静态定义并初始化自旋锁	-	-	无
spin_lock_init(spinlock_t *lock)	void	动态初始化自旋锁(只初始化，并不会分配内存)	-	-	无
spin_lock(spinlock_t *lock)	void	获取自旋锁	阻塞（自旋）	不改变	无
spin_lock_bh(spinlock_t *lock)	void	获取自旋锁并禁止 softirq	阻塞	禁用 softirq	无
spin_lock_irq(spinlock_t *lock)	void	获取自旋锁并禁止本地 CPU 中断	阻塞	禁止本地中断	无
spin_lock_irqsave(spinlock_t *lock, unsigned long flags)	void	获取自旋锁并保存本地中断状态	阻塞	禁止中断，保存 flags	无
spin_trylock(spinlock_t *lock)	int	尝试获取自旋锁	非阻塞	不改变	成功返回 1，失败返回 0
spin_trylock_bh(spinlock_t *lock)	int	尝试获取自旋锁并禁止 softirq	非阻塞	禁用 softirq	成功返回 1，失败返回 0
spin_trylock_irq(spinlock_t *lock)	int	尝试获取自旋锁并禁止中断	非阻塞	禁用本地中断	成功返回 1，失败返回 0
spin_trylock_irqsave(spinlock_t *lock, unsigned long flags)	int	尝试获取自旋锁并保存中断状态	非阻塞	禁用中断，保存 flags	成功返回 1，失败返回 0
spin_unlock(spinlock_t *lock)	void	释放自旋锁	-	-	无
spin_unlock_bh(spinlock_t *lock)	void	释放自旋锁并恢复 softirq 状态	-	恢复 softirq	无
spin_unlock_irq(spinlock_t *lock)	void	释放自旋锁并恢复中断	-	恢复中断	无
spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags)	void	释放自旋锁并恢复中断状态	-	恢复中断状态 flags	无
spin_is_locked(spinlock_t *lock)	int	检查锁是否被持有	非阻塞	-	被锁返回 1，否则 0
spin_is_contended(spinlock_t *lock)	int	检查锁是否存在竞争	非阻塞	-	有竞争返回 1，否则 0

使用自旋锁分为以下 3 步:

1. 在访问临界资源的时候先申请自旋锁
2. 获取到自旋锁之后就进入临界区，获取不到自旋锁就原地等待。
3. 退出临界区的时候要释放自旋锁。

例子

#include <linux/init.h>
#include <linux/module.h>
#include <linux/cdev.h>
#include <linux/slab.h>
#include <linux/spinlock.h>
#include <linux/sched.h>

struct test_drv_data {
        dev_t dev_num;
        struct cdev cdev;
        struct class *class;
        struct device *dev;
};

static struct test_drv_data *test_drv_dat;
static bool status = 1;
static spinlock_t lock;

int spinlock_test_open(struct inode *inode, struct file *file)
{
        file->private_data = test_drv_dat;
        spin_lock(&lock);
        if(status == 0){
                spin_unlock(&lock);
                return -EBUSY;
        }
        status = 0;
        spin_unlock(&lock);
        pr_info("spinlock_test_open is called by [pid: %d]\n",task_pid_nr(current));

        return 0;
}
ssize_t spinlock_test_read(struct file *file, char __user *buf, size_t size, loff_t *offset)
{
        pr_info("spinlock_test_read is called\n");
        return 0;
}
ssize_t spinlock_test_write(struct file *file, const char __user *buf, size_t size, loff_t *offset)
{
        pr_info("spinlock_test_write is called\n");
        return 0;
}
int spinlock_test_release(struct inode *inode, struct file *file)
{
        spin_lock(&lock);
        status = 1;
        spin_unlock(&lock);
        pr_info("spinlock_test_release is called by [pid: %d]\n", task_pid_nr(current));
        return 0;
}
static struct file_operations fops = {
        .owner = THIS_MODULE,
        .open = spinlock_test_open,
        .read = spinlock_test_read,
        .write = spinlock_test_write,
        .release = spinlock_test_release,
};

static int __init spinlock_test_init(void)
{
        int err;
        test_drv_dat = kzalloc(sizeof(struct test_drv_data), GFP_KERNEL);
        if (test_drv_dat == NULL) {
                err = -ENOMEM;
                goto kzalloc_fail;
        }
        err = alloc_chrdev_region(&test_drv_dat->dev_num, 0, 1, "spinlock test chrdev region\n");

        if (err < 0)
                goto alloc_chrdev_region_fail;
        cdev_init(&test_drv_dat->cdev, &fops);
        test_drv_dat->cdev.owner = THIS_MODULE;
        err = cdev_add(&test_drv_dat->cdev, test_drv_dat->dev_num, 1);
        if (err < 0)
                goto cdev_add_fail;
        test_drv_dat->class = class_create(THIS_MODULE, "chrdev");
        if (IS_ERR(test_drv_dat->class)) {
                err = PTR_ERR(test_drv_dat->class);
                goto class_create_fail;
        }
        test_drv_dat->dev = device_create(test_drv_dat->class, NULL, test_drv_dat->dev_num, NULL,
                                          "spinlock_test%d", 0);
        if (IS_ERR(test_drv_dat->dev)) {
                err = PTR_ERR(test_drv_dat->dev);
                goto device_create_fail;
        }
        // 初始化自旋锁
        spin_lock_init(&lock);

        return 0;

device_create_fail:
        class_destroy(test_drv_dat->class);
class_create_fail:
        cdev_del(&test_drv_dat->cdev);
cdev_add_fail:
        unregister_chrdev_region(test_drv_dat->dev_num, 1);
alloc_chrdev_region_fail:
        kfree(test_drv_dat);
kzalloc_fail:
        return err;
}

static void __exit spinlock_test_exit(void)
{
        device_destroy(test_drv_dat->class, test_drv_dat->dev_num);
        class_destroy(test_drv_dat->class);
        cdev_del(&test_drv_dat->cdev);
        unregister_chrdev_region(test_drv_dat->dev_num, 1);
        kfree(test_drv_dat);
}

module_init(spinlock_test_init);
module_exit(spinlock_test_exit);

MODULE_LICENSE("GPL");
MODULE_AUTHOR("even629<asqwgo@163.com>");
MODULE_DESCRIPTION("This is a test sample for spinlock");

测试：

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>

int main(int argc, char **argv)
{
        pid_t pid;
        int fd = -1;

        pid = fork();
        if (pid < 0) {
                perror("fork error");
                fprintf(stderr, "Error: %s[errno: %d]\n", strerror(errno), errno);
                exit(EXIT_FAILURE);
        } else if (pid == 0) { // child process
                printf("[child: <pid: %d>] try to open /dev/spinlock_test0\n",getpid());
                fd = open("/dev/spinlock_test0", O_RDWR);
                if (fd < 0) {
                        perror("[child] open error");
                        fprintf(stderr, "[child] Error: %s[errno: %d]\n", strerror(errno), errno);
                        exit(EXIT_FAILURE);
                }
                printf("[child: <pid: %d>] open /dev/spinlock_test0 success\n", getpid());
                sleep(2);
                close(fd);

        } else { // parent process
                printf("[parent: <pid: %d>] try to open /dev/spinlock_test0\n", getpid());
                fd = open("/dev/spinlock_test0", O_RDWR);
                if (fd < 0) {
                        perror("[parent] open error");
                        fprintf(stderr, "[parent] Error: %s[errno: %d]\n", strerror(errno), errno);
                        exit(EXIT_FAILURE);
                }
                sleep(2);
                printf("[parent: <pid: %d>] open /dev/spinlock_test0 success\n", getpid());

                close(fd);
        }

        return 0;
}

测试用例：

$ insmod spinlock_test.ko
[   10.948417] spinlock_test: loading out-of-tree module taints kernel.
$ ./test_spinlock.o
[child: <pid: 102>] try to open /dev/spinlock_test0
[parent: <pid: 101>] try to open /dev/spinlock_test0
[   16.331206] spinlock_test_open is called by [pid: 102]
[child: <pid: 102>] open /dev/spinlock_test0 success
[parent] open error: Device or resource busy
[parent] Error: Device or resource busy[errno: 16]
$ [   18.335915] spinlock_test_release is called by [pid: 102]

注意事项

1. 由于自旋锁会"原地等待"，因为原地等待会继续占用 CPU，会消耗 CPU 资源，所以锁的时间不能太长。也就是临界区的代码不能太多
2. 在自旋锁保护的临界区里面不能调用可能会导致线程休眠的函数，否则可能会发生死锁
3. 自旋锁一般是用在多核 SOC 上
4. 一次只能有一个任务持有互斥锁；这其实不是规则，而是事实。
5. 多次解锁是不允许的。
6. 它们必须通过API初始化。
7. 持有互斥锁的任务不可能退出，因为互斥锁将保持锁定，可能的竞争者会永远等待（将睡眠）。
8. 不能释放锁定的内存区域。
9. 持有的互斥锁不得重新初始化。
10. 由于它们涉及重新调度，因此互斥锁不能用在原子上下文中，如Tasklet和定时器。

自旋锁死锁

自旋锁死锁是指多个任务或进程因互相等待彼此释放资源，导致谁也无法继续往下执行的现象。

举个例子，假设有进程 A 和进程 B 俩个进程，进程 A 对资源 1 持有自旋锁，同时进程 A 也想获取资源 2。而进程 B 对资源 2 持有自旋锁，同时进程 B 也想获取资源 1。在这种情况下进程 A 和进程 B 都在等待彼此释放资源，从而造成了死锁。

又如进程 A 在拥有自旋锁期间发生了中断，CPU 转头去执行中断处理函数，而此时的中断处理函数也需要获取同一把自旋锁。这时由于锁已被 A 占用，中断处理函数只能等待自旋，从而导致系统进入死锁状态。

• 如在中断处理函数中也要获得自旋锁时，驱动程序则需要在拥有自旋锁时禁止中断（spin_lock_irqsave,spin_lock_irqrestore），在释放自旋锁时使能中断。

• 同时尽量短时间拥有自旋锁，长时间拥有自旋锁可能会导致系统资源耗尽从而造成死锁。

• 最后也要避免某个获得自旋锁的函数调用其他同样试图获取这个锁的函数，否则也会造成死锁。

• 临界区中不能调用任何能够引起休眠或阻塞的函数

读写锁

定义

使用 spinlock 保护临界区时，多个读之间无法并发，只能被 spin，为了提高系统的整体性能，内核定义了一种锁：

• 允许多个处理器进程（或者线程或者中断上下文）并发的进行读操作（SMP 上），这样是安全的，并且提高了 SMP 系统的性能。
• 在写的时候，保证临界区的完全互斥。

读/写自旋锁是在保护 SMP 体系下的共享数据结构而引入的，它的引入是为了增加内核的并发能力。只要内核控制路径没有对数据结构进行修改，读/写自旋锁就允许多个内核控制路径同时读同一数据结构。

如果一个内核控制路径想对这个结构进行写操作，那么它必须首先获取读/写锁的写锁，写锁授权独占访问这个资源。这样设计的目的，即允许对数据结构并发读可以提高系统性能。

例子

/*
 * example_rwlock.c
 */
#include <linux/module.h>
#include <linux/printk.h>
#include <linux/rwlock.h>

static DEFINE_RWLOCK(myrwlock);

static void example_read_lock(void)
{
    unsigned long flags;

    read_lock_irqsave(&myrwlock, flags);
    pr_info("Read Locked\n");

    /* Read from something */

    read_unlock_irqrestore(&myrwlock, flags);
    pr_info("Read Unlocked\n");
}

static void example_write_lock(void)
{
    unsigned long flags;

    write_lock_irqsave(&myrwlock, flags);
    pr_info("Write Locked\n");

    /* Write to something */

    write_unlock_irqrestore(&myrwlock, flags);
    pr_info("Write Unlocked\n");
}

static int __init example_rwlock_init(void)
{
    pr_info("example_rwlock started\n");

    example_read_lock();
    example_write_lock();

    return 0;
}

static void __exit example_rwlock_exit(void)
{
    pr_info("example_rwlock exit\n");
}

module_init(example_rwlock_init);
module_exit(example_rwlock_exit);

MODULE_DESCRIPTION("Read/Write locks example");
MODULE_LICENSE("GPL");

API

初始化 API：

函数	描述
DEFINE_RWLOCK(rwlock_t lock)	定义并初始化读写锁
void rwlock_init(rwlock_t *lock)	初始化读写锁

读操作 API ：

函数	描述
void read_lock(rwlock_t *lock)	获取读锁
void read_unlock(rwlock_t *lock)	释放读锁
void read_lock_irq(rwlock_t *lock)	禁止本地中断，并且获取读锁
void read_unlock_irq(rwlock_t *lock)	打开本地中断，并且释放读锁
void read_lock_irqsave(rwlock_t *lock, unsigned long flags)	保存中断状态，禁止本地中断，并获取读锁
void read_unlock_irqrestore(rwlock_t *lock, unsigned long flags)	将中断状态恢复到以前的状态，并且激活本地 中断，释放读锁
void read_lock_bh(rwlock_t *lock)	关闭下半部，并获取读锁
void read_unlock_bh(rwlock_t *lock)	打开下半部，并释放读锁

写操作 API ：

函数	描述
void write_lock(rwlock_t *lock)	获取写锁
void write_unlock(rwlock_t *lock)	释放写锁
void write_lock_irq(rwlock_t *lock)	禁止本地中断，并且获取写锁
void write_unlock_irq(rwlock_t *lock)	打开本地中断，并且释放写锁
void write_lock_irqsave(rwlock_t *lock, unsigned long flags)	保存中断状态，禁止本地中断，并获取写锁
void write_unlock_irqrestore(rwlock_t *lock, unsigned long flags)	将中断状态恢复到以前的状态，并且激活本地 中断，释放读锁
void write_lock_bh(rwlock_t *lock)	关闭下半部，并获取读锁
void write_unlock_bh(rwlock_t *lock)	打开下半部，并释放读锁

信号量

自旋锁是通过“原地等待”的方式来处理并发与竞争的，所以被保护的临界区不能太长，以免造成对 CPU 资源的浪费。但是有些情况下我们必不可免的要长时间对一些资源进行保护。这时候可以使用信号量。

信号量会引起调用者睡眠，所以信号量也叫睡眠锁。

信号量具有 P 操作和 V 操作。P 操作会使信号量的值减一，如果此时信号量的值小于等于 0，说明资源都被占用，如果还有其他进程需要使用，则该进程会被加入等待队列，如果大于 0，表示可用资源为 n

定义

include/linux/semaphore

/* SPDX-License-Identifier: GPL-2.0-only */
/*
 * Copyright (c) 2008 Intel Corporation
 * Author: Matthew Wilcox <willy@linux.intel.com>
 *
 * Please see kernel/locking/semaphore.c for documentation of these functions
 */
#ifndef __LINUX_SEMAPHORE_H
#define __LINUX_SEMAPHORE_H

#include <linux/list.h>
#include <linux/spinlock.h>

/* Please don't access any members of this structure directly */
struct semaphore {
	raw_spinlock_t		lock;
	unsigned int		count;
	struct list_head	wait_list;
};

#define __SEMAPHORE_INITIALIZER(name, n)				\
{									\
	.lock		= __RAW_SPIN_LOCK_UNLOCKED((name).lock),	\
	.count		= n,						\
	.wait_list	= LIST_HEAD_INIT((name).wait_list),		\
}

#define DEFINE_SEMAPHORE(name)	\
	struct semaphore name = __SEMAPHORE_INITIALIZER(name, 1)

static inline void sema_init(struct semaphore *sem, int val)
{
	static struct lock_class_key __key;
	*sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem, val);
	lockdep_init_map(&sem->lock.dep_map, "semaphore->lock", &__key, 0);
}

extern void down(struct semaphore *sem);
extern int __must_check down_interruptible(struct semaphore *sem);
extern int __must_check down_killable(struct semaphore *sem);
extern int __must_check down_trylock(struct semaphore *sem);
extern int __must_check down_timeout(struct semaphore *sem, long jiffies);
extern void up(struct semaphore *sem);

#endif /* __LINUX_SEMAPHORE_H */

API

API / 宏	返回值类型	功能说明
DEFINE_SEMAPHORE(name)	N/A	定义并初始化一个信号量，初值为 1（互斥信号量）。
sema_init(struct semaphore *sem, int val)	void	初始化一个信号量 sem，初值为 val。
down(struct semaphore *sem)	void	获取信号量。如果计数 > 0，减 1 并立即返回；如果计数为 0，阻塞等待直到可获取。
down_interruptible(struct semaphore *sem)	int	获取信号量。如果被中断信号打断，返回非 0；否则阻塞直到获取成功并返回 0。
down_killable(struct semaphore *sem)	int	获取信号量，如果进程接收到致命信号则返回非 0；否则阻塞直到获取成功并返回 0。
down_trylock(struct semaphore *sem)	int	尝试立即获取信号量。如果获取成功返回非 0，否则立即返回 0，不阻塞。
down_timeout(struct semaphore *sem, long jiffies)	int	尝试获取信号量，最多阻塞 jiffies 个时钟周期。返回值 >0 表示成功，0 表示超时，负值表示错误或信号打断。
up(struct semaphore *sem)	void	释放信号量（计数 +1），并唤醒等待队列中的任务。

down 和 down_interruptible 的区别：

特性	down(struct semaphore *sem)	down_interruptible(struct semaphore *sem)
阻塞行为	如果信号量 count 为 0，调用进程会无限阻塞，直到信号量可用	如果信号量 count 为 0，调用进程会阻塞，但可被中断信号打断
信号打断	不响应信号，中断不会让 down() 返回	响应信号，如果阻塞期间收到信号，则立即返回非 0
返回值	void（永远成功，除非内核 BUG）	int，0 表示获取成功，非 0 表示被信号打断，获取失败
使用场景	不需要响应中断，确保获取信号量	需要可中断阻塞，例如用户进程可以响应 Ctrl+C 或其他信号

示例

#include <linux/init.h>
#include <linux/module.h>
#include <linux/cdev.h>
#include <linux/slab.h>
#include <linux/semaphore.h>
#include <linux/sched.h>

struct test_drv_data {
        dev_t dev_num;
        struct cdev cdev;
        struct class *class;
        struct device *dev;
};

static struct test_drv_data *drv_dat;
static struct semaphore sema;

int semaphore_test_open(struct inode *inode, struct file *file)
{
        int ret;
        file->private_data = drv_dat;
        ret = down_interruptible(&sema);
        if (ret != 0) {
                pr_info("semaphore_test_open is called by [pid: %d] but is interrupted\n",
                        current->pid);
                return -EINTR;
        }
        pr_info("semaphore_test_open is called by [pid: %d]\n", current->pid);
        return 0;
}

int semaphore_test_release(struct inode *inode, struct file *file)
{
        up(&sema);
        pr_info("semaphore_test_release is called by [pid: %d]\n", current->pid);
        return 0;
}
struct file_operations fops = {
        .owner = THIS_MODULE,
        .open = semaphore_test_open,
        .release = semaphore_test_release,
};

static int __init semaphore_test_init(void)
{
        int err;

        drv_dat = (struct test_drv_data *)kzalloc(sizeof(struct test_drv_data), GFP_KERNEL);
        if (drv_dat == NULL) {
                err = -ENOMEM;
                goto kzalloc_fail;
        }

        err = alloc_chrdev_region(&drv_dat->dev_num, 0, 1, "semaphore_test_chrdev_region");
        if (err < 0)
                goto alloc_chrdev_region_fail;
        cdev_init(&drv_dat->cdev, &fops);
        drv_dat->cdev.owner = THIS_MODULE;
        err = cdev_add(&drv_dat->cdev, drv_dat->dev_num, 1);
        if (err < 0)
                goto cdev_add_fail;
        drv_dat->class = class_create(THIS_MODULE, "chrdev_test");
        if (IS_ERR(drv_dat->class)) {
                err = PTR_ERR(drv_dat->class);
                goto class_create_fail;
        }

        drv_dat->dev =
                device_create(drv_dat->class, NULL, drv_dat->dev_num, NULL, "semaphore_test%d", 0);
        if (IS_ERR(drv_dat->dev)) {
                err = PTR_ERR(drv_dat->dev);
                goto device_create_fail;
        }
        // 初始化信号量
        sema_init(&sema, 1);

        return 0;
device_create_fail:
        class_destroy(drv_dat->class);
class_create_fail:
        cdev_del(&drv_dat->cdev);
cdev_add_fail:
        unregister_chrdev_region(drv_dat->dev_num, 1);
alloc_chrdev_region_fail:
        kfree(drv_dat);
kzalloc_fail:
        return err;
}

static void __exit semaphore_test_exit(void)
{
        device_destroy(drv_dat->class, drv_dat->dev_num);
        class_destroy(drv_dat->class);
        cdev_del(&drv_dat->cdev);
        unregister_chrdev_region(drv_dat->dev_num, 1);
        kfree(drv_dat);
}

module_init(semaphore_test_init);
module_exit(semaphore_test_exit);

MODULE_LICENSE("GPL");
MODULE_AUTHOR("even629<asqwgo@163.com>");
MODULE_DESCRIPTION("This is test sample for semaphore");

测试用例：

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <unistd.h>

int main(int argc, char **argv)
{
        pid_t pid;
        int fd;

        pid = fork();
        if (pid < 0) {
                perror("fork error");
                exit(EXIT_FAILURE);
        } else if (pid == 0) { // child
                fd = open("/dev/semaphore_test0", O_RDWR);
                if (fd < 0) {
                        perror("child open /dev/semaphore_test0 error");
                        exit(EXIT_FAILURE);
                }
                printf("[child: pid<%d>] open /dev/semaphore_test0 success\n", getpid());
                sleep(2);
                close(fd);
        } else { // parent
                fd = open("/dev/semaphore_test0", O_RDWR);
                if (fd < 0) {
                        perror("parent open /dev/semaphore_test0 error");
                        exit(EXIT_FAILURE);
                }
                printf("[parent: pid<%d>] open /dev/semaphore_test0 success\n", getpid());
                sleep(2);
                close(fd);
        }

        return 0;
}

$ insmod semaphore_test.ko
[   15.981179] semaphore_test: loading out-of-tree module taints kernel.
$ ./test_semaphore.o
[   20.535287] semaphore_test_open is called by [pid: 101]
[parent: pid<101>] open /dev/semaphore_test0 success
[   22.544668] semaphore_test_release is called by [pid: 101]
[   22.545409] semaphore_test_open is called by [pid: 102]
[child: pid<102>] open /dev/semaphore_test0 success
[   24.550326] semaphore_test_release is called by [pid: 102]

注意

1. 信号量的值不能小于 0
2. 访问共享资源时，信号量执行"减一"操作，访问完成后再执行"加一"操作
3. 当信号量的值为 0，想访问共享资源的线程必须等待，当信号量大于 0 时，等待的线程才可以访问
4. 因为信号量会引起休眠，所以中断里面不能使用信号量
5. 如果共享资源持有时间比较长，一般用信号量而不用自旋锁
6. 在同时使用信号量和自旋锁的时候，要先获取信号量，再使用自旋锁。因为信号量会导致睡眠。

读写信号量

与读写锁的区别在于读写信号量是睡眠锁，而读写锁是自旋锁，选择：

临界区是否可能睡眠？
├─ 是 → 使用读写信号量 (rw_semaphore) 或 互斥锁 (mutex)
│ └─ 读操作远多于写操作？→ 是 → 读写信号量
│ └─ 读操作不显著多于写操作？→ 互斥锁更简单，性能更好
│
└─ 否 → 临界区是否非常短暂（几个整数操作）？
├─ 是 → 普通自旋锁 (spinlock_t) 可能比读写自旋锁更好
└─ 否 → 读远多于写？→ 是 → 读写自旋锁 (rwlock_t)
└─ 否 → 普通自旋锁

中断上下文中不能使用可能导致睡眠的函数：

**因为中断上下文不属于任何进程，它没有进程控制块（task_struct），因此从根本上不具备“睡眠”所需的基础设施。**当一个进程要进入睡眠状态时，它会主动将自己标记为“睡眠”状态，从运行队列移出，并告诉调度器：“请在锁可用时叫醒我。” 这个操作依赖进程的上下文。

API

API函数定义	功能说明
DECLARE_RWSEM(name)	声明名为name的读写信号量，并初始化它。
void init_rwsem(struct rw_semaphore *sem);	对读写信号量sem进行初始化。
void down_read(struct rw_semaphore *sem);	读者用来获取sem，若没获得时，则调用者睡眠等待。
void up_read(struct rw_semaphore *sem);	读者释放sem。
int down_read_trylock(struct rw_semaphore *sem);	读者尝试获取sem，如果获得返回1，如果没有获得返回0。可在中断上下文使用。
void down_write(struct rw_semaphore *sem);	写者用来获取sem，若没获得时，则调用者睡眠等待。
int down_write_trylock(struct rw_semaphore *sem);	写者尝试获取sem，如果获得返回1，如果没有获得返回0。可在中断上下文使用
void up_write(struct rw_semaphore *sem);	写者释放sem。
void downgrade_write(struct rw_semaphore *sem);	把写者降级为读者。

示例

初始化：

#include <linux/rwsem.h>

/* 静态声明并初始化 */
DECLARE_RWSEM(my_rwsem);

/* 动态初始化（结构体成员） */
struct rw_semaphore my_rwsem;
init_rwsem(&my_rwsem);

使用读写锁：

/* 读者：读取配置并拷贝到用户空间 */
ssize_t my_read(struct file *filp, char __user *buf, size_t size, loff_t *off)
{
    int ret;
    
    down_read(&my_rwsem);                    /* 获取读锁，可睡眠 */
    ret = copy_to_user(buf, my_data, size);  /* 允许，因为持有可能睡眠的信号量 */
    up_read(&my_rwsem);                      /* 释放读锁 */
    
    return ret;
}

/* 写者：更新配置 */
ssize_t my_write(struct file *filp, const char __user *buf, size_t size, loff_t *off)
{
    down_write(&my_rwsem);                   /* 获取写锁，独占 */
    copy_from_user(my_data, buf, size);      /* 允许睡眠 */
    up_write(&my_rwsem);                     /* 释放写锁 */
    
    return size;
}

互斥锁

同一个资源同一个时间只有一个访问者在进行访问，其他的访问者结束以后才可以访问这个资源。这就是互斥。

互斥锁和信号量为 1 的情况很类似，但是互斥锁更简洁高效。不过要注意的事项更多。

定义

include/linux/mutex.h

struct mutex {
	atomic_long_t		owner;
	spinlock_t		wait_lock;
#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
	struct optimistic_spin_queue osq; /* Spinner MCS lock */
#endif
	struct list_head	wait_list;
#ifdef CONFIG_DEBUG_MUTEXES
	void			*magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
	struct lockdep_map	dep_map;
#endif
};

API

函数/宏	返回值类型	功能说明
DEFINE_MUTEX(name)	struct mutex	静态定义并初始化一个互斥锁，初始状态为 unlocked
mutex_init(struct mutex *lock)	void	初始化互斥锁为 unlocked 状态
mutex_destroy(struct mutex *lock)	void	销毁互斥锁（仅在 DEBUG_MUTEXES 下实际工作）
mutex_lock(struct mutex *lock)	void	获取互斥锁，如果锁被占用则阻塞直到锁可用
mutex_lock_interruptible(struct mutex *lock)	int	获取互斥锁，如果锁被占用则阻塞，可被信号中断，成功返回 0，被信号打断返回 -EINTR
mutex_lock_killable(struct mutex *lock)	int	获取互斥锁，如果锁被占用则阻塞，可被致死信号打断，返回值同上
mutex_lock_io(struct mutex *lock)	void	获取互斥锁，允许在 I/O 上下文使用（不常用）
mutex_lock_nested(struct mutex *lock, unsigned int subclass)	void	嵌套锁获取，用于 Lockdep 分析，实际等效于 mutex_lock()
mutex_lock_interruptible_nested(struct mutex *lock, unsigned int subclass)	int	可中断的嵌套锁获取，用于 Lockdep 分析
mutex_lock_killable_nested(struct mutex *lock, unsigned int subclass)	int	可致死信号打断的嵌套锁获取，用于 Lockdep 分析
mutex_lock_nest_lock(struct mutex *lock, struct lockdep_map *nest_lock)	void	嵌套锁获取，用于 Lockdep 分析
mutex_trylock(struct mutex *lock)	int	尝试获取互斥锁，不阻塞，成功返回 1，失败返回 0
mutex_trylock_recursive(struct mutex *lock)	enum mutex_trylock_recursive_enum	尝试获取互斥锁，允许递归，返回 0/1/递归标志
mutex_unlock(struct mutex *lock)	void	释放互斥锁，必须由持有锁的线程调用
atomic_dec_and_mutex_lock(atomic_t *cnt, struct mutex *lock)	int	将原子计数器减 1，如果减到 0 则获取 mutex；否则返回 0
mutex_is_locked(struct mutex *lock)	bool	查询互斥锁是否被占用，true 表示锁被持有，false 表示未被持有

与等待队列的可中断系列函数一样，建议使用mutex_lock_interruptible()，它使驱动程序可以被所有信号中断，而对于mutex_lock_killable()，只有杀死进程的信号才能中断驱动程序。

调用mutex_lock()时要非常小心，只有能够保证无论在什么情况下互斥锁都会释放时才可以使用它。在用户上下文中，建议始终使用mutex_lock_interruptible() 来获取互斥锁，因为mutex_lock() 即使收到信号（甚至是Ctrl+C组合键），也不会返回。

示例

#include <linux/init.h>
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/cdev.h>
#include <linux/mutex.h>
#include <linux/sched.h>

struct test_drv_data {
        dev_t dev_num;
        struct cdev cdev;
        struct class *class;
        struct device *dev;
};

static struct test_drv_data *drv_dat;
static struct mutex drv_mutex;

int mutex_test_open(struct inode *inode, struct file *file)
{
        int err;
        file->private_data = drv_dat;
        err = mutex_lock_interruptible(&drv_mutex);
        if (err != 0) {
                pr_err("pid: %d is interrupted", current->pid);
                return err;
        }
        pr_info("mutex_test_open is called by pid: %d\n", current->pid);
        return 0;
}

int mutex_test_release(struct inode *inode, struct file *file)
{
        mutex_unlock(&drv_mutex);
        pr_info("mutex_test_release is called by pid: %d\n", current->pid);
        return 0;
}

static struct file_operations fops = {
        .owner = THIS_MODULE,
        .open = mutex_test_open,
        .release = mutex_test_release,
};

static int __init mutex_test_init(void)
{
        int err;
        drv_dat = (struct test_drv_data *)kzalloc(sizeof(struct test_drv_data), GFP_KERNEL);

        if (drv_dat == NULL) {
                err = -ENOMEM;
                goto kzalloc_fail;
        }

        err = alloc_chrdev_region(&drv_dat->dev_num, 0, 1, "mutex_test_chrdev_region");
        if (err < 0)
                goto alloc_chrdev_region_fail;

        cdev_init(&drv_dat->cdev, &fops);
        drv_dat->cdev.owner = THIS_MODULE;
        err = cdev_add(&drv_dat->cdev, drv_dat->dev_num, 1);
        if (err < 0)
                goto cdev_add_fail;

        drv_dat->class = class_create(THIS_MODULE, "chrdev_test");
        if (IS_ERR(drv_dat->class)) {
                err = PTR_ERR(drv_dat->class);
                goto class_create_fail;
        }

        drv_dat->dev =
                device_create(drv_dat->class, NULL, drv_dat->dev_num, NULL, "mutex_test%d", 0);
        if (IS_ERR(drv_dat->dev)) {
                err = PTR_ERR(drv_dat->dev);
                goto device_create_fail;
        }
        // 初始化互斥锁
        mutex_init(&drv_mutex);
        return 0;
device_create_fail:
        class_destroy(drv_dat->class);
class_create_fail:
        cdev_del(&drv_dat->cdev);
cdev_add_fail:
        unregister_chrdev_region(drv_dat->dev_num, 1);
alloc_chrdev_region_fail:
        kfree(drv_dat);
kzalloc_fail:
        return err;
}

static void __exit mutex_test_exit(void)
{
        kfree(drv_dat);
}

module_init(mutex_test_init);
module_exit(mutex_test_exit);

MODULE_LICENSE("GPL");
MODULE_AUTHOR("even629<asqwgo@163.com>");
MODULE_DESCRIPTION("This is a test sample for mutex");

测试用例：

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <unistd.h>

int main(int argc, char **argv)
{
        pid_t pid;
        int fd;
        char prefix[16];        

        pid = fork();
        if (pid < 0) {
                perror("fork error");
                exit(EXIT_FAILURE);
        } else if (pid == 0) {
                snprintf(prefix, sizeof(prefix), "[pid: %d]", getpid());

                fd = open("/dev/mutex_test0", O_RDWR);
                if (fd < 0) {
                        perror(prefix);
                        exit(EXIT_FAILURE);
                }
                printf("%s child open success\n", prefix);
                sleep(2);
                close(fd);

        } else {
                snprintf(prefix, sizeof(prefix), "[pid: %d]", getpid());

                fd = open("/dev/mutex_test0", O_RDWR);
                if (fd < 0) {
                        perror(prefix);
                        exit(EXIT_FAILURE);
                }
                printf("%s parent open success\n", prefix);
                sleep(2);
                close(fd);
        }

        return 0;
}

$ insmod mutex_test.ko
[   13.589343] mutex_test: loading out-of-tree module taints kernel.
$ ./test_mutex.o
[   17.141253] mutex_test_open is called by pid: 101
[pid: 101] parent open success
[   19.150144] mutex_test_release is called by pid: 101
[   19.151379] mutex_test_open is called by pid: 102
[pid: 102] child open success
[   21.156199] mutex_test_release is called by pid: 102

注意事项

1. 互斥锁会导致休眠，因此在中断中不能使用互斥锁
2. 同一时刻只能有一个线程持有互斥锁，并且只有持有者可以解锁
3. 不允许递归上锁和解锁

RCU

RCU（Read-Copy-Update，读-复制-更新） 作为一种高性能的并发控制机制，专为解决“读多写少”场景而设计。它通过分离读操作与写操作的同步逻辑，使读操作几乎无开销（无需加锁、无原子指令、无阻塞），而写操作则通过“复制后更新”和“延迟释放”的方式保证安全性。

RCU 已成为 Linux 内核中不可或缺的基础组件，广泛应用于进程调度、内存管理、文件系统等核心模块。

RCU的适用场景主要有以下特点：

1. 数据结构主要是通过指针访问;
2. 读取操作远多于更新操作;
3. 读取端代码不能容忍锁的开销;
4. 更新操作相对不频繁;

RCU的核心思想可以概括为**“读时直接读, 写时拷贝改”**

举一个生活场景：家里有一块公告板, 上面记录着家庭成员的日程安排. 当妈妈需要更新日程时, 她不会阻止其他人查看公告板, 而是先将当前内容抄到一张新纸上, 在新纸上修改, 然后在所有人都不会注意的瞬间, 将新纸替换到公告板上. 替换后, 她并不立即销毁旧纸张, 而是会等待一段时间, 确保所有可能在看旧公告板的人都看完后, 才将旧纸张丢弃

在这个比喻中, “替换瞬间” 对应RCU中的发布（Publishing） 操作, “等待确保” 对应宽限期（Grace Period）, 而**“丢弃旧纸”** 则对应回收（Reclamation）. 这三个概念构成了RCU的核心三角

术语

• Read-Side Critical Section（读侧临界区）：读操作访问 RCU 保护数据的代码段，需用 rcu_read_lock() 和 rcu_read_unlock() 标记。在此区间内，读操作可安全访问旧数据（即使写操作已更新数据）。

• Grace Period（宽限期）：从写操作开始更新数据到所有读侧临界区（针对旧数据的）结束的时间段。宽限期结束意味着“所有可能访问旧数据的读操作均已完成”，此时旧数据可安全释放。

• Write-Side Operation（写侧操作）：包含三个步骤：

  1. 复制（Copy）：创建数据的副本（或新数据）。
  2. 更新（Update）：原子地将指针从旧数据切换到新数据。
  3. 延迟释放（Deferred Free）：等待宽限期结束后，释放旧数据。

RCU 工作原理

Read-Side

读侧操作的无锁访问是 RCU 性能优势的核心来源。其流程如下：

1. 通过 rcu_read_lock() 进入读侧临界区（本质是禁用抢占，或在 Preemptible RCU 中跟踪读侧状态）。
2. 使用 rcu_dereference(p) 获取指向共享数据的指针（包含内存屏障，确保指针读取的原子性和可见性）。
3. 访问指针指向的数据（无需加锁，直接读取）。
4. 通过 rcu_read_unlock() 退出读侧临界区。

rcu_read_lock();
/* 读取受保护的数据 */
data = rcu_dereference(global_pointer);
/* 使用数据 */
rcu_read_unlock();

这里的 rcu_read_lock() 和 rcu_read_unlock() 并不像传统锁那样进行实际的加锁和解锁操作. 在非抢占式内核中, 它们可能完全是空操作. 它们的真正作用是告诉编译器不要重排指令, 并在可抢占内核中标记临界区的开始和结束

rcu_dereference() 是一个关键的宏, 它确保在Alpha等弱内存顺序的处理器上, 读取操作能够按预期顺序进行. 它可以看作是内存屏障的一种封装, 确保指针的读取在获取实际数据之前发生

读取端的关键特性在于零开销 - 没有原子操作, 没有内存屏障（在大多数架构上）, 没有锁争用。这使得RCU读取路径极其高效, 即使在高并发读取场景下, 性能也能保持稳定。

Write-Side

写侧操作需保证“旧数据在所有读操作完成后释放”，流程如下：

1. 复制/创建新数据：例如，为链表节点分配新内存并初始化。
2. 原子更新指针：使用 rcu_assign_pointer(p, new_ptr) 将共享指针从旧数据切换到新数据（包含内存屏障，确保新指针对所有 CPU 可见）。
3. 等待宽限期：通过 synchronize_rcu()（阻塞等待）或 call_rcu()（异步回调）等待宽限期结束。
4. 释放旧数据：宽限期结束后，安全释放旧数据（如 kfree(old_ptr)）。

例子：

// 写侧操作：更新 global_ptr
struct data *new_data = kmalloc(sizeof(*new_data), GFP_KERNEL);
if (!new_data)
   	return -ENOMEM;
// 初始化新数据...
new_data->value = new_value;

// 发布操作，原子更新指针，对读侧可见
rcu_assign_pointer(global_ptr, new_data);

// 等待宽限期结束（所有旧数据的读操作已完成）
synchronize_rcu();

// 释放旧数据
kfree(old_data);

Grace Period

如何保证安全释放？

宽限期是 RCU 的核心机制，其作用是确保所有在“指针更新前”开始的读侧临界区均已结束。RCU 如何检测宽限期结束？

• 核心思想：每个 CPU 维护一个“RCU 状态”（如 rcu_seq），记录当前是否有活跃的读侧临界区。当写操作请求宽限期时，RCU 会等待所有 CPU 经历一次“上下文切换”（如调度、中断返回等）——这意味着该 CPU 上的读侧临界区已结束（因为读侧临界区禁用抢占，无法被调度打断，发生上下文切换了就说明Read-Side已经结束）。
• 实现细节：Linux 内核通过 rcu_node 树状结构跟踪各 CPU 的 RCU 状态，synchronize_rcu() 会阻塞直到所有 CPU 都完成一次“宽限期确认”（rcu_qs()）。

宽限期可能长达数百毫秒（取决于系统负载），写侧不应假设其快速结束，避免设计依赖宽限期时长的逻辑。

Linux内核通过上下文切换、用户模式执行和空闲循环作为读取端临界区退出的标志. 当一个CPU经历了这些状态之一后, 内核认为该CPU上的所有RCU读取端临界区都已经完成

RCU的实现严重依赖于内存屏障来保证多核环境下的顺序一致性. 防止编译器和CPU为了优化而重排指令

• 在读取端, rcu_dereference()包含了读内存屏障, 确保在获取指针之后的操作不会重排到指针获取之前.

• 在更新端, rcu_assign_pointer()包含了写内存屏障, 确保在更新指针之前的所有存储操作对其它CPU可见之前, 指针更新本身不会对其它CPU可见

这种顺序保证对于RCU的正确性至关重要. 考虑数据发布的场景：更新者必须先初始化新数据, 然后发布指针. 如果没有内存屏障, CPU或编译器可能会重排这两步, 导致读取者看到未完全初始化的数据

RCU 同步原语

Read-Side

void rcu_read_lock(void);

void rcu_read_unlock(void);

以上两个API用于标记读侧临界区的开始与结束，确保在此区间内读操作可安全访问旧数据。

• 在 Classic RCU（非抢占 RCU）中，rcu_read_lock() 禁用抢占（preempt_disable()），rcu_read_unlock() 启用抢占（preempt_enable()）。
• 在 Preemptible RCU（可抢占 RCU）中，通过 rcu_read_lock_bh() 禁用软中断，或使用 rcu_read_lock() 配合内核抢占计数。

可以用rcu_dereference来解引用

typeof(p) rcu_dereference(p);

读侧安全访问 RCU 保护的指针，确保：

• 指针读取的原子性（防止编译器优化导致的部分指针加载）。
• 内存屏障（确保新数据对读侧可见，避免 CPU 缓存不一致）。

示例：

// 读侧操作：访问 RCU 保护的指针
rcu_read_lock();
struct data *d = rcu_dereference(global_ptr); // 安全获取指针
if (d)
   	printk("value: %d\n", d->value); // 访问数据
rcu_read_unlock();

Write-Side

rcu_assign_pointer

void rcu_assign_pointer(p, typeof(p) v);

写侧原子更新 RCU 保护的指针，确保：

• 指针更新的原子性。
• 内存屏障（确保新数据完全初始化后才对读侧可见）。

synchronize_rcu()

• 功能：阻塞等待宽限期结束，返回后旧数据可安全释放。

• 注意：不能在原子上下文（如中断处理函数）中使用（会导致调度失败）。

• 示例：

  // 删除 RCU 保护的链表节点
  list_del_rcu(node); // 从链表中移除节点（RCU 安全版本）
  synchronize_rcu();  // 等待宽限期结束
  kfree(node);        // 释放旧节点

synchronize_rcu() 会阻塞等待宽限期，原子上下文（如中断、软中断）无法调度，导致内核崩溃。此时应使用 call_rcu()。

call_rcu()

• 功能：异步等待宽限期，宽限期结束后调用指定的回调函数释放旧数据（非阻塞）。

• 原型：void call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *head));

• 示例：

  // 定义回调函数（释放数据）
  void free_data(struct rcu_head *head) {
  	struct data *d = container_of(head, struct data, rcu);
  	kfree(d);
  } 
  
  // 写侧操作：更新数据并异步释放旧数据
  struct data *new_data = kmalloc(...);
  struct data *old_data = rcu_dereference_protected(global_ptr, ...);
  rcu_assign_pointer(global_ptr, new_data); // 更新指针
  call_rcu(&old_data->rcu, free_data);      // 宽限期后调用 free_data

• 读侧必须用 rcu_dereference()：禁止直接访问指针（如 d = global_ptr），否则可能因编译器优化或 CPU 乱序导致数据不一致。
• 写侧必须用 rcu_assign_pointer()：禁止直接赋值（如 global_ptr = new_d），确保新指针对所有 CPU 可见。

RCU 使用示例

RCU 保护的指针更新

场景：全局指针 global_data 指向一个结构体，读操作频繁访问，写操作偶尔更新。

#include <linux/rcupdate.h>
#include <linux/sched.h>
 
// 定义共享数据结构（含 RCU 头，用于 call_rcu）
struct my_data {
    int value;
    struct rcu_head rcu; // 必须包含，用于 call_rcu 回调
};
 
// RCU 保护的全局指针
static struct my_data __rcu *global_data;
 
// 读侧函数：访问数据
void read_data(void) {
    struct my_data *d;
 
    rcu_read_lock();
    d = rcu_dereference(global_data); // 安全获取指针
    if (d)
        printk("Read value: %d\n", d->value);
    rcu_read_unlock();
}
 
// 写侧函数：更新数据（使用 call_rcu 异步释放）
void update_data(int new_val) {
    struct my_data *new_d, *old_d;
 
    new_d = kmalloc(sizeof(*new_d), GFP_KERNEL);
    if (!new_d)
        return;
    new_d->value = new_val;
 
    // 原子更新指针（读侧可见）
    old_d = rcu_dereference_protected(global_data, 1); // 写侧安全获取旧指针
    rcu_assign_pointer(global_data, new_d);
 
    // 宽限期后释放旧数据
    if (old_d)
        call_rcu(&old_d->rcu, (void (*)(struct rcu_head *))kfree);
}

RCU 保护的链表操作

RCU 常用于保护动态链表（如进程列表、网络连接表）。内核提供了 list_add_rcu()、list_del_rcu() 等 RCU 安全的链表操作宏。

场景：维护一个 RCU 保护的双向链表，读侧遍历链表，写侧添加/删除节点。

#include <linux/list.h>
#include <linux/rculist.h> // RCU 链表宏
 
// 链表节点结构
struct my_node {
    int id;
    struct list_head list;
    struct rcu_head rcu;
};
 
// RCU 保护的链表头
static LIST_HEAD(my_list);
static DEFINE_SPINLOCK(list_lock); // 写侧互斥锁（多个写者需同步）
 
// 读侧：遍历链表（无锁）
void traverse_list(void) {
    struct my_node *node;
 
    rcu_read_lock();
    // list_for_each_entry_rcu：RCU 安全的链表遍历宏
    list_for_each_entry_rcu(node, &my_list, list) {
        printk("Node id: %d\n", node->id);
    }
    rcu_read_unlock();
}
 
// 写侧：添加节点
void add_node(int id) {
    struct my_node *node = kmalloc(sizeof(*node), GFP_KERNEL);
    if (!node)
        return;
    node->id = id;
    INIT_LIST_HEAD(&node->list);
 
    // 写侧需加锁（防止多个写者同时修改链表）
    spin_lock(&list_lock);
    list_add_rcu(&node->list, &my_list); // RCU 安全的添加
    spin_unlock(&list_lock);
}
 
// 写侧：删除节点
void delete_node(int id) {
    struct my_node *node, *tmp;
 
    spin_lock(&list_lock);
    list_for_each_entry_safe(node, tmp, &my_list, list) {
        if (node->id == id) {
            list_del_rcu(&node->list); // RCU 安全的删除
            spin_unlock(&list_lock);
            // 等待宽限期后释放节点
            call_rcu(&node->rcu, (void (*)(struct rcu_head *))kfree);
            return;
        }
    }
    spin_unlock(&list_lock);
}

RCU 最佳实践

最小化 Read-Side 临界区长度

读侧临界区越长，宽限期可能越长（写侧需等待更久），导致旧数据延迟释放，增加内存压力。

原则：仅在访问 RCU 数据时进入临界区，避免在临界区内执行耗时操作（如 I/O、复杂计算）。

// 错误示例：临界区包含耗时操作
rcu_read_lock();
d = rcu_dereference(global_data);
if (d) {
    heavy_computation(); // 耗时操作，不应在临界区内
    printk("%d\n", d->value);
}
rcu_read_unlock();
 
// 正确示例：缩小临界区
rcu_read_lock();
d = rcu_dereference(global_data);
if (d)
    value = d->value; // 仅读取必要数据
rcu_read_unlock();
heavy_computation(); // 临界区外执行
printk("%d\n", value);

RCU 变种

Linux 内核提供多种 RCU 变种，适用于不同场景：

• Classic RCU（CONFIG_RCU_FANOUT）：默认变种，不支持读侧抢占，适用于非实时内核。
• Preemptible RCU（CONFIG_PREEMPT_RCU）：支持读侧抢占，读侧临界区可被调度，适用于实时内核。
• Sleepable RCU（SRCU，CONFIG_SRCU）：读侧可阻塞（如调用 msleep()），需用 srcu_read_lock()/srcu_read_unlock()，适用于读侧需睡眠的场景。

PER_CPU

示例

net/ipv6/af_inet6.c

static int __net_init ipv6_init_mibs(struct net *net)
{
	int i;

	net->mib.udp_stats_in6 = alloc_percpu(struct udp_mib);
	if (!net->mib.udp_stats_in6)
		return -ENOMEM;
	net->mib.udplite_stats_in6 = alloc_percpu(struct udp_mib);
	if (!net->mib.udplite_stats_in6)
		goto err_udplite_mib;
	net->mib.ipv6_statistics = alloc_percpu(struct ipstats_mib);
	if (!net->mib.ipv6_statistics)
		goto err_ip_mib;

	for_each_possible_cpu(i) {
		struct ipstats_mib *af_inet6_stats;
		af_inet6_stats = per_cpu_ptr(net->mib.ipv6_statistics, i);
		u64_stats_init(&af_inet6_stats->syncp);
	}


	net->mib.icmpv6_statistics = alloc_percpu(struct icmpv6_mib);
	if (!net->mib.icmpv6_statistics)
		goto err_icmp_mib;
	net->mib.icmpv6msg_statistics = kzalloc(sizeof(struct icmpv6msg_mib),
						GFP_KERNEL);
	if (!net->mib.icmpv6msg_statistics)
		goto err_icmpmsg_mib;
	return 0;

err_icmpmsg_mib:
	free_percpu(net->mib.icmpv6_statistics);
err_icmp_mib:
	free_percpu(net->mib.ipv6_statistics);
err_ip_mib:
	free_percpu(net->mib.udplite_stats_in6);
err_udplite_mib:
	free_percpu(net->mib.udp_stats_in6);
	return -ENOMEM;
}

struct net 中的 struct netns_mib mib 成员定义如下：

struct netns_mib {
	DEFINE_SNMP_STAT(struct tcp_mib, tcp_statistics);
	DEFINE_SNMP_STAT(struct ipstats_mib, ip_statistics);
	DEFINE_SNMP_STAT(struct linux_mib, net_statistics);
	DEFINE_SNMP_STAT(struct udp_mib, udp_statistics);
	DEFINE_SNMP_STAT(struct udp_mib, udplite_statistics);
	DEFINE_SNMP_STAT(struct icmp_mib, icmp_statistics);
	DEFINE_SNMP_STAT_ATOMIC(struct icmpmsg_mib, icmpmsg_statistics);

#if IS_ENABLED(CONFIG_IPV6)
	struct proc_dir_entry *proc_net_devsnmp6;
	DEFINE_SNMP_STAT(struct udp_mib, udp_stats_in6);
	DEFINE_SNMP_STAT(struct udp_mib, udplite_stats_in6);
	DEFINE_SNMP_STAT(struct ipstats_mib, ipv6_statistics);
	DEFINE_SNMP_STAT(struct icmpv6_mib, icmpv6_statistics);
	DEFINE_SNMP_STAT_ATOMIC(struct icmpv6msg_mib, icmpv6msg_statistics);
#endif
#ifdef CONFIG_XFRM_STATISTICS
	DEFINE_SNMP_STAT(struct linux_xfrm_mib, xfrm_statistics);
#endif
#if IS_ENABLED(CONFIG_TLS)
	DEFINE_SNMP_STAT(struct linux_tls_mib, tls_statistics);
#endif
#ifdef CONFIG_MPTCP
	DEFINE_SNMP_STAT(struct mptcp_mib, mptcp_statistics);
#endif
};

而宏DEFINE_SNMP_STAT定义如下：

#define DEFINE_SNMP_STAT(type, name)	\
	__typeof__(type) __percpu *name

参考文献

[1]

Linux RCU 深度剖析：原理、实践与最佳实践

[2]

Linux RCU 机制深入解析：原理与实践

[3]

What is RCU

[4]

What is RCU Fundamentally?

[5]

Linux 驱动开发 二十八：读写锁

[6]

Per-cpu变量