07-设备驱动中的并发控制

7.1 并发与竞态并发concurrency指多个执行路径可能同时访问同一份数据竞态race condition指这些访问的先后顺序会影响最终结果。Linux 内核中的并发来源比 4.0 时代更多读驱动时必须逐一列出并发源说明典型场景多进程多个用户进程同时read()/write()/ioctl()同一个设备两个进程同时写 globalmem多线程同一进程内多线程共享 fd 与文件私有数据线程池写同一个字符设备软中断/任务tasklet、net_rx_action、timer 回调、workqueue定时器回调修改设备状态硬中断设备产生 IRQ随时打断进程上下文按键中断更新计数内核抢占CONFIG_PREEMPT*下进程上下文可被抢占PREEMPT_RT上几乎处处可抢占SMP 多核真正并行执行双核 QEMU-smp 2异步 I/O / io_uring多个 in-flight 请求并行提交io_uring 批量提交RCU 回调 / 软中断延迟执行回调在另一时刻运行call_rcu()竞态的本质是临界区没有互斥保护。驱动开发中的通用流程是找出所有可能并发访问的共享数据设备结构体、全局变量、寄存器。判断每个访问点的上下文进程/软中断/硬中断/原子上下文。选择能在该上下文中使用的保护机制。明确锁的顺序避免死锁。7.1.1 上下文判断表先判断再选锁当前上下文可以睡眠可以用 mutex可以用 spinlock可用copy_to_user()进程上下文是是是是进程上下文持有 spinlock否否是否软中断 / tasklet否否是否硬中断顶半部否否是否threaded IRQ 线程是是是是中断被禁止的临界区否否是否判断依据是内核的in_interrupt()、in_softirq()、in_atomic()、preempt_count()。7.2.5 中in_atomic()的语义见include/linux/preempt.hmight_sleep()会在运行时检查不该睡眠的地方睡眠。7.2 编译乱序和执行乱序原书用一整节讲这个问题7.2.5 下依然成立而且因为编译器优化更激进风险更大。编译乱序编译器在-O2下可以重排无依赖的内存访问。执行乱序CPU 的 store buffer、乱序执行、cache 一致性协议会导致另一核观察到的写入顺序与程序顺序不同。内核提供的工具工具头文件用途READ_ONCE()/WRITE_ONCE()include/asm-generic/rwonce.h禁止编译器把一次访问拆开或合并barrier()include/linux/compiler.h仅阻止编译器重排smp_mb()/smp_rmb()/smp_wmb()include/asm-generic/barrier.h全/读/写内存屏障dma_wmb()/dma_rmb()include/asm-generic/barrier.hDMA 场景的屏障smp_load_acquire()/smp_store_release()include/linux/atomic.h获取/释放语义atomic_t的_acquire/_release变体include/linux/atomic/带序原子操作RCU_INIT_POINTER/rcu_assign_pointerinclude/linux/rcupdate.h发布指针经验法则只要数据可能被无锁读例如被poll读、被中断读、被另一核读写侧就要用WRITE_ONCE()或原子/锁读侧用READ_ONCE()。发布指针 数据时用rcu_assign_pointer()smp_store_release()语义消费侧用rcu_dereference()。设备寄存器读写用readl()/writel()它们自带__iomem的barrier()语义见include/asm-generic/io.h。7.3 中断屏蔽最原始的保护方式只在极短的临界区内使用。unsigned long flags; local_irq_save(flags); /* 关本核中断并保存状态 */ /* ... 临界区 ... */ local_irq_restore(flags); /* 恢复 */ local_irq_disable(); /* 关本核中断不保存状态不推荐嵌套 */ local_irq_enable(); local_bh_disable(); /* 只关软中断 */ local_bh_enable();7.2.5 的变化local_irq_save()依旧但推荐改用spin_lock_irqsave()因为关中断代价高且 4.0 之后的PREEMPT_RT会把 spinlock 转成可睡眠锁手工关中断的语义在 RT 上不一致。local_irq_save()只作用于当前 CPU不能保护多核并行。preempt_disable()/preempt_enable()是另一层禁止抢占而非关中断。7.4 原子操作7.4.1 整型原子操作/* include/linux/atomic.h 与 include/linux/atomic/atomic-arch-fallback.h */ atomic_t v ATOMIC_INIT(0); atomic_inc(v); /* v */ atomic_dec(v); /* v-- */ atomic_add(i, v); /* v i */ atomic_sub(i, v); /* v - i */ atomic_read(v); /* 读可能被合并需配合内存序 */ atomic_set(v, i); /* 写 */ /* 返回旧值 */ int old atomic_fetch_add(i, v); int old atomic_fetch_inc(v); int old atomic_fetch_sub(i, v); /* 带序变体 */ atomic_inc_return(v); atomic_dec_and_test(v); /* 减到 0 返回 true */ atomic_add_negative(i, v); /* 加后为负返回 true */ atomic_inc_and_test(v); /* 7.2.5 推荐带内存序的接口 */ atomic_add_return_acquire(i, v); atomic_add_return_release(i, v); atomic_add_return_relaxed(i, v);atomic64_t提供同样接口atomic64_inc()等在 32 位平台上可能由锁实现。注意atomic_t只能保护一个整数不能保护多个字段的一致性。例如引用计数为 0 时释放结构体必须用refcount_tinclude/linux/refcount.h它会在检测到 0→1 的复活时触发refcount_warn_saturate()。refcount_t r REFCOUNT_INIT(1); refcount_inc(r); /* 加引用 */ if (refcount_dec_and_test(r)) /* 减到 0 时释放 */ kfree(obj);7.4.2 位原子操作/* include/asm-generic/bitops/atomic.h, include/linux/bitops.h */ unsigned long flags 0; set_bit(nr, flags); /* 置位 */ clear_bit(nr, flags); /* 清位 */ change_bit(nr, flags); /* 翻转 */ test_bit(nr, flags); /* 测试非原子语义上仍是原子的读 */ test_and_set_bit(nr, flags); /* 置位并返回旧值 */ test_and_clear_bit(nr, flags); /* 清位并返回旧值 */ test_and_change_bit(nr, flags); /* 非原子的位操作必须自己加锁 */ __set_bit(nr, flags); __clear_bit(nr, flags);7.2.5 新增/常用接口test_bit()现在在部分架构上通过arch_test_bit()实现且带_acquire变体。bitmap_*()系列include/linux/bitmap.h用于大位图如bitmap_set()、bitmap_clear()、bitmap_weight()。assign_bit()与test_and_set_bit_lock()/clear_bit_unlock()用于锁位模式。7.5 自旋锁自旋锁用于不能睡眠的上下文中断、软中断、原子上下文或临界区极短的场景。自旋期间当前 CPU 忙等。7.5.1 自旋锁的使用/* include/linux/spinlock.h */ spinlock_t lock; /* 定义 */ spin_lock_init(lock); /* 动态初始化 */ DEFINE_SPINLOCK(lock); /* 静态初始化 */ spin_lock(lock); /* ... 临界区不能睡眠不能调用 copy_to_user() ... */ spin_unlock(lock); /* 进程上下文 中断上下文共享必须关中断 */ unsigned long flags; spin_lock_irqsave(lock, flags); /* ... */ spin_unlock_irqrestore(lock, flags); /* 明确知道中断已关 */ spin_lock_irq(lock); spin_unlock_irq(lock); /* 只关软中断 */ spin_lock_bh(lock); spin_unlock_bh(lock);选择规则原书表格的 7.2.5 版本数据被谁访问进程上下文使用说明只有进程上下文且不睡眠持有spin_lock()需要禁止抢占进程 软中断spin_lock_bh()防止本核软中断插入进程 硬中断spin_lock_irqsave()防止本核中断插入进程 硬中断 多核spin_lock_irqsave()多核由自旋本身保证7.2.5 的重要变化裸spin_lock()在CONFIG_PREEMPT_RT下变成可睡眠的rt_mutex因此RT 内核里不要在原子上下文持裸 spinlock应改用raw_spinlock_t。自旋锁不可递归同一线程重复spin_lock()会死锁。spin_trylock()/spin_trylock_irqsave()提供非阻塞获取可用于poll等快速返回路径。spin_lock_irqsave()的返回值语义在 RT 上被保留但flags的用途不同必须成对使用。7.5.2 读写自旋锁允许多个读者并发、写者独占。/* include/linux/rwlock.h */ rwlock_t rwlock; rwlock_init(rwlock); DEFINE_RWLOCK(rwlock); read_lock(rwlock); read_unlock(rwlock); write_lock(rwlock); write_unlock(rwlock); /* 带中断版本 */ read_lock_irqsave(rwlock, flags); read_unlock_irqrestore(rwlock, flags); write_lock_irq(rwlock); write_unlock_irq(rwlock);7.2.5 的建议读写自旋锁的读者会阻塞写者造成写者饥饿且 RT 上表现不佳。新代码优先用 RCU 或 seqlock读写自旋锁只用于极短临界区。7.5.3 顺序锁seqlock 允许读者无锁读、写者独占写读者可能读到不一致的数据需要重试。/* include/linux/seqlock.h */ seqlock_t lock; seqlock_init(lock); DEFINE_SEQLOCK(lock); /* 写者 */ write_seqlock(lock); /* ... 更新数据 ... */ write_sequnlock(lock); /* 读者 */ unsigned int seq; do { seq read_seqbegin(lock); /* ... 读取数据可能不一致 ... */ } while (read_seqretry(lock, seq));7.2.5 相关接口read_seqbegin_or_lock()/need_seqretry()/done_seqretry()读者在多次重试后升级为持锁读避免极端情况下活锁。seqcount_t不带锁的纯序列计数用于jiffies_64读取、struct timespec读取等场景。seqcount_latch_tinclude/linux/seqlock.h用于需要快照稳定的场景。7.5.4 读-复制-更新RCURCU 是 4.0 之后驱动中最重要的无锁技术之一适合读多写少的指针发布场景。/* include/linux/rcupdate.h */ struct my_obj { struct rcu_head rcu; int value; }; struct my_obj __rcu *global_obj; /* 读侧可在软中断/中断上下文使用不能睡眠 */ rcu_read_lock(); p rcu_dereference(global_obj); if (p) use(p-value); rcu_read_unlock(); /* 写侧发布新对象 */ new kmalloc(...); rcu_assign_pointer(global_obj, new); /* 写侧等待所有读者离开后释放旧对象 */ synchronize_rcu(); kfree(old); /* 或在 RCU 回调中释放不能在原子上下文用 synchronize_rcu */ call_rcu(old-rcu, my_free_cb);7.2.5 的 RCU 家族变体用途rcu_read_lock()/synchronize_rcu()标准 RCU可被抢占rcu_read_lock_bh()软中断保护rcu_read_lock_sched()调度器 RCUsrcu_read_lock()/synchronize_srcu()可睡眠读侧sleepable RCUrcu_read_lock_trace()tracing 用rcu_tasks任务级 RCU用于 trampoline 等要点RCU 读侧临界区不能睡眠SRCU 除外。rcu_dereference()保证读到完整的指针含地址依赖屏障。释放对象必须用kfree_rcu()、call_rcu()或synchronize_rcu()之一绝不能在rcu_assign_pointer()后立刻kfree()。7.2.5 中kfree_rcu()支持双参数宏kfree_rcu(ptr, rhf)。7.6 信号量信号量是可睡眠的计数器锁4.0 时代用于互斥和同步。7.2.5 中/* include/linux/semaphore.h */ struct semaphore sem; sema_init(sem, 1); DEFINE_SEMAPHORE(sem); /* 计数为 1 */ down(sem); /* 不可中断获取 */ down_interruptible(sem); /* 可被信号打断返回 -EINTR */ down_killable(sem); down_trylock(sem); /* 非阻塞返回非 0 表示失败 */ up(sem); /* 释放 */7.2.5 的现实down()家族仍存在但新代码几乎不应使用计数信号量需要互斥 → 用mutex有所有权、有 lockdep 检查、可调试。需要最多 N 个并发 → 用completion、sema_init(sem, N)仍是合理选择但更推荐struct completion 计数逻辑或atomic_t。需要读者写者 → 用rw_semaphoredown_read()/down_write()或 RCU。信号量没有所有者概念一个线程可以down()而另一个线程up()这正是它与 mutex 的本质区别。7.7 互斥体/* include/linux/mutex.h */ struct mutex lock; mutex_init(lock); DEFINE_MUTEX(lock); mutex_lock(lock); /* ... 临界区可以睡眠 ... */ mutex_unlock(lock); /* 可中断版本 */ if (mutex_lock_interruptible(lock)) return -ERESTARTSYS; /* 可杀死版本 */ mutex_lock_killable(lock); /* 非阻塞 */ if (mutex_trylock(lock)) mutex_unlock(lock);7.2.5 要点mutex有所有者只能由获取者释放违反会触发DEBUG_MUTEXES警告。mutex不可递归。mutex不能在中断/软中断/原子上下文使用会might_sleep()报警。mutex 适合长时间持有的临界区短临界区用 spinlock。mutex_lock()在CONFIG_DEBUG_MUTEXES下会做死锁检测配合 lockdep。7.2.5 的include/linux/mutex.h中mutex_lock()已改为__mutex_lock()的封装参数含mutex_acquire等 lockdep 钩子。scope-based 清理7.2.5 新风格#include linux/cleanup.h static int foo(struct my_dev *dev) { guard(mutex)(dev-lock); /* 进入作用域即加锁离开自动解锁 */ if (bad_condition(dev)) return -EINVAL; /* 自动解锁不需要 goto */ return 0; }include/linux/cleanup.h提供guard()、scoped_guard()、DEFINE_GUARD()、DEFINE_FREE()、__free()等。这是 4.0 时代完全没有的机制能显著减少goto err_unlock样板代码。7.8 完成量完成量用于一个线程等待另一件事完成是同步而非互斥。/* include/linux/completion.h */ struct completion done; init_completion(done); DECLARE_COMPLETION(done); DECLARE_COMPLETION_ONSTACK(done); /* 等待方 */ wait_for_completion(done); /* 不可中断 */ wait_for_completion_interruptible(done); /* 可中断 */ wait_for_completion_killable(done); wait_for_completion_timeout(done, HZ); /* 超时 */ wait_for_completion_interruptible_timeout(done, msecs_to_jiffies(100)); wait_for_completion_io(done); /* 不计入 IO 等待统计的变体 */ /* 唤醒方可在中断上下文调用 complete */ complete(done); /* 唤醒一个等待者 */ complete_all(done); /* 唤醒全部 */ /* 复用 */ reinit_completion(done);7.2.5 新增wait_for_completion_state()可指定等待状态。complete_all()之后done.done为UINT_MAX/2reinit_completion()会重置。有completion_acquire()/completion_release()的 lockdep 支持见include/linux/completion.h。典型用法驱动probe()提交固件加载请求中断处理函数complete()唤醒等待。7.9 增加并发控制后的 globalmem 设备驱动原书用 globalmem 演示不加锁会怎样。下面给出 7.2.5 下推荐的两种写法。7.9.1 写法一mutex 显式错误路径兼容老代码风格#include linux/module.h #include linux/fs.h #include linux/cdev.h #include linux/device.h #include linux/mutex.h #include linux/uaccess.h #define GLOBALMEM_SIZE 0x1000 #define MEM_CLEAR 0x1 static int major; static struct class *globalmem_class; struct globalmem_dev { struct cdev cdev; struct device *device; struct mutex lock; /* 保护 mem 与 current_pos */ unsigned char mem[GLOBALMEM_SIZE]; unsigned int current_pos; }; static struct globalmem_dev *globalmem_devp; static int globalmem_open(struct inode *inode, struct file *filp) { struct globalmem_dev *dev container_of(inode-i_cdev, struct globalmem_dev, cdev); filp-private_data dev; return 0; } static ssize_t globalmem_read(struct file *filp, char __user *buf, size_t count, loff_t *ppos) { struct globalmem_dev *dev filp-private_data; unsigned long p *ppos; int ret; if (p GLOBALMEM_SIZE) return 0; if (count GLOBALMEM_SIZE - p) count GLOBALMEM_SIZE - p; if (mutex_lock_interruptible(dev-lock)) return -ERESTARTSYS; if (copy_to_user(buf, dev-mem p, count)) { ret -EFAULT; } else { *ppos count; ret count; pr_info(globalmem: read %zu bytes from %lu\n, count, p); } mutex_unlock(dev-lock); return ret; } static ssize_t globalmem_write(struct file *filp, const char __user *buf, size_t count, loff_t *ppos) { struct globalmem_dev *dev filp-private_data; unsigned long p *ppos; int ret; if (p GLOBALMEM_SIZE) return 0; if (count GLOBALMEM_SIZE - p) count GLOBALMEM_SIZE - p; if (mutex_lock_interruptible(dev-lock)) return -ERESTARTSYS; if (copy_from_user(dev-mem p, buf, count)) { ret -EFAULT; } else { *ppos count; ret count; pr_info(globalmem: written %zu bytes from %lu\n, count, p); } mutex_unlock(dev-lock); return ret; } static loff_t globalmem_llseek(struct file *filp, loff_t offset, int orig) { loff_t ret; switch (orig) { case SEEK_SET: ret offset; break; case SEEK_CUR: ret filp-f_pos offset; break; case SEEK_END: ret GLOBALMEM_SIZE offset; break; default: return -EINVAL; } if (ret 0 || ret GLOBALMEM_SIZE) return -EINVAL; filp-f_pos ret; return ret; } static long globalmem_ioctl(struct file *filp, unsigned int cmd, unsigned long arg) { struct globalmem_dev *dev filp-private_data; switch (cmd) { case MEM_CLEAR: if (mutex_lock_interruptible(dev-lock)) return -ERESTARTSYS; memset(dev-mem, 0, GLOBALMEM_SIZE); dev-current_pos 0; mutex_unlock(dev-lock); pr_info(globalmem: cleared\n); return 0; default: return -EINVAL; } } static const struct file_operations globalmem_fops { .owner THIS_MODULE, .open globalmem_open, .read globalmem_read, .write globalmem_write, .llseek globalmem_llseek, .unlocked_ioctl globalmem_ioctl, }; static int __init globalmem_init(void) { int ret, i; dev_t devno; ret alloc_chrdev_region(devno, 0, 1, globalmem); if (ret) return ret; major MAJOR(devno); globalmem_devp kzalloc(sizeof(*globalmem_devp), GFP_KERNEL); if (!globalmem_devp) { ret -ENOMEM; goto err_region; } mutex_init(globalmem_devp-lock); cdev_init(globalmem_devp-cdev, globalmem_fops); globalmem_devp-cdev.owner THIS_MODULE; ret cdev_add(globalmem_devp-cdev, devno, 1); if (ret) goto err_free; globalmem_class class_create(globalmem); /* 7.2.5只有一个参数 */ if (IS_ERR(globalmem_class)) { ret PTR_ERR(globalmem_class); goto err_cdev; } globalmem_devp-device device_create(globalmem_class, NULL, devno, NULL, globalmem); if (IS_ERR(globalmem_devp-device)) { ret PTR_ERR(globalmem_devp-device); goto err_class; } pr_info(globalmem: loaded, major%d\n, major); return 0; err_class: class_destroy(globalmem_class); err_cdev: cdev_del(globalmem_devp-cdev); err_free: kfree(globalmem_devp); err_region: unregister_chrdev_region(devno, 1); return ret; } static void __exit globalmem_exit(void) { dev_t devno MKDEV(major, 0); device_destroy(globalmem_class, devno); class_destroy(globalmem_class); cdev_del(globalmem_devp-cdev); kfree(globalmem_devp); unregister_chrdev_region(devno, 1); pr_info(globalmem: unloaded\n); } module_init(globalmem_init); module_exit(globalmem_exit); MODULE_LICENSE(GPL); MODULE_AUTHOR(manual); MODULE_DESCRIPTION(globalmem with mutex, Linux 7.2.5);7.9.2 写法二7.2.5 风格cdev_device_addguard#include linux/cleanup.h static int globalmem_probe_one(struct globalmem_dev *dev) { int ret; ret alloc_chrdev_region(dev-devno, 0, 1, globalmem); if (ret) return ret; mutex_init(dev-lock); cdev_init(dev-cdev, globalmem_fops); dev-cdev.owner THIS_MODULE; dev-device device_create(globalmem_class, NULL, dev-devno, dev, globalmem%d, dev-idx); if (IS_ERR(dev-device)) { ret PTR_ERR(dev-device); goto err_region; } /* cdev_device_add() 会调用 device_add()因此必须先把 device 加到类里 * 或者让 cdev 的 parent 指向已注册 device。这里演示 cdev_add 的等价写法。 */ ret cdev_add(dev-cdev, dev-devno, 1); if (ret) goto err_device; return 0; err_device: device_destroy(globalmem_class, dev-devno); err_region: unregister_chrdev_region(dev-devno, 1); return ret; }重要提醒7.2.5cdev_device_add()内部会调用device_add()而device_create()也会调用device_add()。两者不能混用否则会重复注册同一个struct device并触发 sysfs 警告。正确做法是二选一推荐device_initialize()cdev_device_add()put_device()简单device_create()cdev_add()device_destroy()cdev_del()。7.9.3 QEMU 验证脚本# 宿主机构建模块 make -C ~/works/study/linux-7.2.5 \ O~/works/study/build/linux-7.2.5-x86_64 \ M$PWD modulesguest 中insmod globalmem.ko ls -l /dev/globalmem cat /sys/class/globalmem/globalmem/dev # 并发写测试 for i in $(seq 1 8); do (echo thread$i /dev/globalmem; ) done; wait dd if/dev/globalmem bs64 count1 2/dev/null | hexdump -C观察点加锁前并发写会出现数据交错加锁后dmesg中written N bytes的偏移量互不重叠。7.9.4 用 lockdep 验证锁的正确性在 QEMU 内核配置中打开CONFIG_PROVE_LOCKINGy CONFIG_DEBUG_LOCK_ALLOCy CONFIG_DEBUG_SPINLOCKy CONFIG_DEBUG_MUTEXESy CONFIG_LOCK_STATy CONFIG_DEBUG_ATOMIC_SLEEPyguest 中mount -t debugfs none /sys/kernel/debug cat /sys/kernel/debug/lockdep # 需要 lockdep 开启 cat /sys/kernel/debug/locking/lockdep_stats echo 1 /sys/kernel/debug/provoke-crash/DIRECT # 可选验证 oops 路径7.10 常见错误错误现象修正在 spinlock 临界区调用copy_to_user()BUG: sleeping function called from invalid context把用户拷贝移到锁外锁内只拷贝到内核缓冲用 mutex 保护中断上下文数据同上或死锁改用 spinlock 或irqsave忘记spin_unlock_irqrestore使用同一个flags中断状态错乱严格成对rcu_assign_pointer()后立即kfree()UAF偶发 oops用kfree_rcu()/synchronize_rcu()在rcu_read_lock()中睡眠scheduling while atomic改用srcu或移到锁外用atomic_t保护多字段结构偶发不一致用 spinlock/mutex自旋锁递归获取死锁单核直接挂起拆锁或用*_bh/*_irq变体用mutex_lock()却不处理-EINTR用户 Ctrl-C 无法打断用mutex_lock_interruptible()忘记reinit_completion()复用完成量第二次等待立即返回复用前reinit_completion()读写锁写者饥饿系统卡顿改 RCU 或 seqlock7.11 本章实验清单实验目的关键观察无锁 globalmem 并发写观察竞态dmesg偏移量重叠、数据被覆盖mutex 版 globalmem观察互斥偏移量串行、数据完整guard(mutex)版学习 7.2.5 清理机制错误路径不泄漏锁lockdep 检测锁序学习死锁检测dmesg中出现possible circular locking dependencyRCU 链表增删学习无锁读rcu_read_lock内不睡眠7.12 总结并发控制的第一步是判断上下文而不是先选锁。atomic_t/refcount_t只解决单变量多字段一致性必须用锁。spinlock 用于不能睡眠的场景mutex 用于可睡眠的长临界区RCU 用于读多写少seqlock 用于读多且能重试。编译乱序和执行乱序必须用READ_ONCE()/WRITE_ONCE()/屏障显式处理。7.2.5 的include/linux/cleanup.hguard()/__free()能显著减少错误路径泄漏。任何锁都必须能被 lockdep 验证在 QEMU 中打开CONFIG_PROVE_LOCKING跑一遍你的驱动。