并发控制

原子变量

原子变量底层是一个整型，但是不能把他当作整型去操作

定义原子变量

//定义一个原子变量，并且赋值
atomic_t flag = ATOMIC_INIT(0);
void atomic_set(flag,0); //设置原子变量的值

获取一个原子变量的值

flag = atomic_read(flag);

原子变量加/减

void atomic_add(int i,flag);  //相当于flag + i
void atomic_sub(int i,flag);  //相当于flag - i

原子变量自增/自减

void atomic_inc(flag); //相当于flag++
void atomic_dec(flag);//相当于flag--

测试原子变量是否正确

对原子变量进行了自加，自减，减法之后再去进行测试

修改变量值，并检查修改后的值是否为0，整个操作过程保证不会被中断（原子性）。

int atomic_inc_and_test(&flag);
int atomic_dec_and_test(&flag);
int atomic_sub_and_test(&flag);

1. `int atomic_inc_and_test(atomic_t *v)`

作用：将变量 v 原子地加1，然后检查新值是否为0。
返回值：如果加1后 v 的值等于 0，则返回 真（非0）；否则返回 假（0）。
使用场景：较少单独使用，通常用于需要计数到零边界触发的场景（例如计数器从 -1 加到 0）。

2. `int atomic_dec_and_test(atomic_t *v)`

作用：将变量 v 原子地减1，然后检查新值是否为0。
返回值：如果减1后 v 的值等于 0，则返回 真（非0）；否则返回 假（0）。
使用场景：这是最常用的函数，主要用于引用计数。当释放一个资源时，计数减1，如果返回真（代表计数变为0），说明没有人在使用该资源，此时可以安全地执行释放内存、关闭文件等清理操作。

3. `int atomic_sub_and_test(int i, atomic_t *v)`

作用：将变量 v 原子地减去 i，然后检查新值是否为0。
返回值：如果减去 i 后 v 的值等于 0，则返回 真（非0）；否则返回 假（0）。
使用场景：批量减少计数，当需要一次性释放多个引用时使用（比如释放一个包含多个子对象的缓存块）。

操作并返回

int atomic_add_return(int i ,flag);
int atomic_sub_return(int i ,flag);
int atomic_inc_return(flag);
int atomic_dec_return(flag);

相当于加减后直接返回值，就不需要上面那样再atomic_read

自旋锁

自旋锁是一种用于保护共享资源的锁机制，特别适用于多处理器或多核环境下的短期资源访问。当一个线程尝试获取锁时，如果锁已被其他线程占用，该线程不会进入睡眠，而是不断循环检查锁的状态，直到锁被释放并成功获取锁

这里不进入睡眠是因为睡眠会造成进程调度，会有上下文切换开销，线程不会进入阻塞状态

定义一个自旋锁

spinlock_t lock;

初始化自旋锁

spin_lock_init(lock);

获取/释放自旋锁

spin_lock(lock);
spin_unlock(lock);

尝试获得一个spin

不阻塞，尝试进行获得锁，成功返回1

int spin_trylock(lock)

衍生函数：spin+irq的操作函数

spin_lock_irq(lock) = spin_lock(lock) + local_irq_disable();//都是当前CPU，不是全部
spin_unlock_irq(lock) = spin_unlock(lock) + local_irq_enable();
spin_lock_irqsave(lock,flags) = spin_lock(lock) + local_irq_save();
spin_unlock_irqrestore(lock,flags) = spin_unlock(lock) + local_irq_restore();

spin使用注意

临界区一定不能太长，要在极短的时间完成，太长会降低系统性能，会造成死锁
临界区不能递归使用同一个spin，某个有了spin的CPU又要获取这个spin，就会造成CPU的死锁
在临界区不能调用可能引起进程调度的函数，会造成临界区阻塞，造成系统崩溃,例如copy_from_user,copy_to_user,kalloc,msleep
要做好跨平台的设计；

例如使用spin_lock_irqsave和spin_unlock_irqrestore保护临界区，这样我当前的核的中断服务函数（isr）就不能调用spin_lock和spin_unlock造成阻塞
但是如果是SMP系统 ，就算当前核使用spin_lock_irqsave和spin_unlock_irqrestore保护临界区，但是其他核的中断服务函数还是可以调用spin_lock和spin_unlock

所以为了当前驱动能够做到跨平台（单核/SMP），不仅要在进程中使用spin_lock_irqsave和spin_unlock_irqrestore,也需要在中断服务函数中使用spin_lock和spin_unlock

自旋锁（Spinlock）的核心作用就是保证互斥。当你成功获取自旋锁后，意味着只有当前CPU核心上的执行流能进入这个临界区，其他CPU核心哪怕想进来，也都在“原地打转”等待锁被释放。

既然临界区内不存在多核并发，那么对普通变量进行 cnt++ 或 cnt-- 这种操作，就不会被其他核心打断修改。
所以，锁本身已经取代了原子操作的必要性。你在锁里用原子操作，等于“穿着防弹衣（锁）还要再拿个盾牌（原子变量）”，既多余又增加开销。

读写锁

读写锁是对spin的一种优化

相比于spin来说读写锁的有点在于可以多个进程去读，只能有一个进程去写

读锁加锁，锁的不是“其他读进程”，而是“写进程”，同时也为了锁住“读进程自身的缓存状态”。

定义读写锁/初始化读写锁

rwlock_t my_rwlock;
rwlock_init(my_rwlock);

读锁定操作函数

void read_lock(rwlock_t *my_rwlock);
void read_lock_irq(rwlock_t *my_rwlock);
void read_lock_irqsave(rwlock_t *my_rwlock,flags);

读解锁

void read_unlock(rwlock_t *my_rwlock);
void read_unlock_irq(rwlock_t *my_rwlock);
void read_unlock_irqrestore(rwlock_t *my_rwlock,flags);

写锁定函数

void write_lock(rwlock_t *my_rwlock);
void write_lock_irq(rwlock_t *my_rwlock);
void write_lock_irqrestore(rwlock_t *my_rwlock,flags);
void write_trylock(rwlock_t *my_rwlock);

顺序锁

顺序锁是读写锁的一种优化

顺序锁相对于读写锁的优点是：可以读写不排斥

顺序锁（Seqlock，即 Sequential Lock）是读写锁的一种极致优化版本，专门针对**“写操作极少，但读操作极其频繁”**的场景（比如系统时间 gettimeofday、网络统计计数）。

虽然读者不阻塞写者，但写者之间是互斥的（通常结合自旋锁）

核心组件：一个整型序列计数器（Sequence Counter）

顺序锁维护一个整型的 seq 计数器（初始值为0）。这个计数器有一个隐含的奇偶规则：

偶数（包括0）：表示当前没有写者在修改数据，数据处于一致状态。
奇数：表示当前有写者正在临界区内部修改数据，数据处于“撕裂”状态。

读者（Reader）的操作流程（核心：重试循环）

读者永远不需要抢锁，也不会阻塞写者，但必须用 do-while 循环来“自证清白”：

// 典型的内核读操作伪代码
do {
 seq = read_seqbegin(&seqlock); // 1. 读取当前版本号（偶数）

// 2. 这里开始拷贝数据（比如读取 64位时间戳）
data = shared_data; 

} while (read_seqretry(&seqlock, seq)); // 3. 检查版本号是否改变

检查逻辑（read_seqretry）：

如果当前 seq 等于刚才读到的 seq（且为偶数）：说明在读数据期间，没有任何写者进来过，数据完整一致，退出循环。
如果当前 seq 不等于刚才读到的 seq（可能是奇数，或者变成了更大的偶数）：说明在读数据期间，写者闯进来并修改完了数据。你刚才拷贝的数据是“新老混杂”的垃圾数据，必须丢掉重来，跳回 do 重新读一遍。

定义一个顺序锁

seqlock_t seqlock;
seqlock_init(seqlock);

获取一个顺序锁

void write_seqlock(seqlock_t *seqlock);
int write_tryseqlock(seqlock_t *seqlock);
void write_seqlock_irq(seqlock_t *seqlock);
void write_seqlock_irqrestore(seqlock_t *seqlock,flags);

读开始函数

unsigned read_seqbegin(const seqlock_t *sl);
read_seqbegin_irqsave(seqlock_t *seqlock,flags);

重读函数

int read_seqretry(const seqlock_t *sl);
read_seqretry_irqrestore(seqlock_t *seqlock,flags);

RCU(read-copy-update)

RCU（Read-Copy-Update，读-复制-更新）是Linux内核中一种非常独特的同步机制

它的核心思想是将更新操作拆分为“移除”和“回收”两个阶段，从而允许读操作与更新操作完全并发地执行

核心原理

RCU的工作流程可以分解为以下三步：

读（Read）：完全无锁，零开销。
读者要访问数据时，只需调用rcu_read_lock()标记临界区的开始。这个操作开销极小，在不可抢占的内核中甚至可能是一条内存屏障指令。读者通过指针读取数据，并保证在rcu_read_unlock()之前完成所有操作。关键约定：在读临界区内，代码不能睡眠或主动让出CPU。
复制（Copy）：写者“秘密”修改。
当写者需要修改数据时，它不会直接在原数据上操作。而是先创建一个新的副本，在副本上完成所有修改。
更新（Update）：“发布”新数据，等待“宽限期”。

发布新数据：修改完成后，写者使用特殊的rcu_assign_pointer() API，原子地将指向共享数据的指针从旧版本切换到新版本。这一步是“发布”，从这一刻起，新的读者将看到新数据。
等待宽限期（Grace Period）：此时，旧数据依然存在，可能还有老读者在访问。写者必须调用synchronize_rcu()等函数，进入等待状态。内核会监测所有CPU，直到确认所有在更新前开始的读操作都已退出临界区。这个等待期被称为“宽限期”。
安全回收：宽限期结束后，意味着没有任何读者会再访问旧数据，写者这时才可以安全地将旧数据占用的内存释放或回收。

顺序锁也是一种针对读多写少优化的机制。它与RCU的核心区别在于优先级和代价：

顺序锁：写者优先。写者不会被读者阻塞，但读者在读取过程中若发现数据被修改，需要重试（Retry）。这会给读者带来额外开销，在写操作频繁时可能导致读者“饿死”。
RCU：读者优先。读者完全无开销，但写者需要承担复制、等待和回收的复杂工作。

API函数

RCU读锁定

rcu_read_lock();

读解锁

rcu_read_unlock();

同步RCU操作(阻塞)

该函数主要用于在写操作完成之后等待所有读操作完成，然后将共享数据指针从旧版本切换到新版本

synchronize_rcu()

非阻塞回调同步RCU

该函数可以非阻塞调用，可以用在中断中

void call_rcu(struct rcu_head *head,void(* func)(start rcu_head *rcu);

给RCU保护的指针赋初值

在链表的写端函数，发布

rcu_assign_pointer(old,new);

确保“新数据结构的初始化”一定发生在“指针被指向新地址”之前，绝对不能让读者看到一个“只改了一半”的垃圾数据。

获取一个rcu保护指针

在链表的读端使用，订阅

rcu_dereference(new);

rcu_assign_pointer和rcu_dereference一般配合使用

插入链表头操作

static inline void list_add_rcu(static list_head *new,list_head *head);

插入链表尾操作

static inline void list_add_tail_rcu(struct list_head *new,list_head *head);;

删除一个rcu保护链表的节点

这里的entry是一个入口，比如某一个结构体里面有一个list_head成员，就可以传递它进入

static inline void list_del_rcu(struct list_head *entry);

新链表元素取代旧链表元素

static inline void list_replace_rcu(struct list_head *old,struct list_head *new)

注意与rcu_assign_pointer的区别

rcu_assign_pointer(p, v)：泛型工具。它操作的是一个独立的指针变量（比如全局指针 g_foo）。它只负责把那一个指针的值从 A 改成 B。
list_replace_rcu(old, new)：专用工具。它操作的是 Linux 内核双链表（struct list_head）中的节点。它不仅要让新节点出现在旧节点的位置上，还要维护链表前后向连接的完整性。

遍历RCU保护链表函数

static inline void list_for_each_rcu(pos,head);

标准操作

我们假设要保护一个全局数据结构 struct foo，它是一个被多核频繁读取、偶尔更新的统计信息。

1. 场景定义（头文件）

#include <linux/rcupdate.h>
#include <linux/slab.h>
struct foo {
 int a;
 int b;
 // 可能还有更多字段
};
// 受 RCU 保护的全局指针，初始为 NULL
struct foo __rcu *g_foo;

2. 读端（Reader）标准写法

读端的特点是：极轻量，但绝对不能睡眠。

void read_foo(void)
{
 struct foo *p;
 int val_a, val_b;
 // 1. 进入 RCU 读临界区（本质是一个禁止抢占的标记）
 // 警告：这里绝对绝对不能调用 schedule()、mutex_lock()、kmalloc(, GFP_KERNEL) 等！
 rcu_read_lock();
 // 2. 获取指针的最新发布版本（带内存屏障，保证看到的是完整初始化的数据）
 p = rcu_dereference(g_foo);
 if (p) {
 // 3. 安全地读取数据（此时即使有写者在更新，也不会影响我们读到的这一份版本）
 val_a = p->a;
 val_b = p->b;
 // 使用 val_a, val_b ...
 }
 // 4. 退出读临界区（恢复抢占）
 rcu_read_unlock();
}

3. 写端（Updater）更新数据写法

写端的特点是：“复制一份，修改，发布指针，等待老读者离开，回收”。

void update_foo(int new_a, int new_b)
{
 struct foo *p_old;
 struct foo *p_new;
 // 1. 申请新内存，复制旧数据（如果旧数据存在）
 // 这里可以用 GFP_KERNEL，允许睡眠，因为不在 RCU 读锁内！
 p_new = kmalloc(sizeof(struct foo), GFP_KERNEL);
 if (!p_new) return;
 // 2. 获取旧指针，并填充新数据
 // 写者之间需要额外加锁！这里假设外界有自旋锁保护写者并发。
 p_old = rcu_dereference_protected(g_foo, lockdep_is_held(&my_lock));
 if (p_old) {
 *p_new = *p_old; // 拷贝旧内容
 }
 p_new->a = new_a;
 p_new->b = new_b;
 // 3. 原子地发布新指针（释放屏障：保证上面初始化先完成，指针后生效）
 rcu_assign_pointer(g_foo, p_new);
 // 4. 等待一个“宽限期（Grace Period）”
 // 让所有在 rcu_assign_pointer 之前进入 rcu_read_lock() 的读者退出临界区。
 synchronize_rcu();
 // 5. 宽限期过后，没有任何读者持有旧指针了，安全释放旧内存
 kfree(p_old);
}

4. 替换（删除）指针的标准写法

如果你想把指针置为 NULL（比如模块卸载）：

void delete_foo(void)
{
 struct foo *p_old;
 // 写者互斥锁保护（假设 my_lock）
 p_old = rcu_dereference_protected(g_foo, lockdep_is_held(&my_lock));
 // 把全局指针置为 NULL（发布）
 rcu_assign_pointer(g_foo, NULL);
 // 等待宽限期
 synchronize_rcu();
 // 安全释放旧内存
 kfree(p_old);
}

注意

一定要先等待宽限期synchronize_rcu再去释放旧的数据

互斥体mutex

互斥体主要用于临界区有上下文切换，临界区很长的情况,所以也适合用于有用户态内核态的数据拷贝的情况

定义和初始化互斥体

struct mutex my_mutex;
mutex_init(&my_mutex);

获取互斥体

void mutex_lock(struct mutex *lock);
int mutex_lock_interruptible(struct mutex *lock);
int mutex_trylock(struct mutex *lock);

释放互斥体

void mutex_unlock(struct mutex *lock);

关于自旋锁和睡眠锁区别

在Linux内核中，是否允许在临界区内调度，是区分“睡眠锁”和“自旋锁”的关键，而不是互斥锁与其他锁（如读写锁、顺序锁）之间的本质区别。

Linux内核的锁主要分为两类，它们在能否睡眠/调度的特性上完全不同：

睡眠锁 (Sleeping Locks)：允许在临界区内睡眠或调度。如果锁被占用，请求锁的进程会主动让出CPU，进入睡眠状态等待。这类锁包括互斥锁(mutex)、信号量(semaphore)、读写信号量(rw_semaphore) 等。
自旋锁 (Spinning Locks)：绝对禁止在临界区内睡眠或调度。如果锁被占用，请求锁的进程会原地“自旋”忙等，直到锁被释放。这类锁包括原始自旋锁(raw_spinlock_t)、自旋锁(spinlock_t)、读写锁(rwlock_t) 等。顺序锁(seqlock_t) 在写者侧也是基于自旋锁实现的，因此同样适用此规则。

自旋锁临界区睡眠的“死局”

自旋锁的核心规则是：持有自旋锁期间，禁止抢占（Preemption）和睡眠。一旦违反，在单核和多核系统上都会陷入死局：

单核 CPU 场景（直接卡死）：
持有锁的任务 A 在临界区内睡眠（主动让出 CPU）。此时，CPU 去执行任务 B。任务 B 恰好也要拿同一把自旋锁，发现锁被占用，于是原地自旋忙等。但此时整个系统只有一个 CPU，而持有锁的任务 A 处于睡眠状态，永远得不到 CPU 来运行并释放锁。结果就是：B 永远在自旋，A 永远睡不醒，系统彻底卡死（死锁）。
多核 CPU 场景（软锁死 Lockup）：
任务 A 在 CPU 0 上持有锁并睡眠。任务 B 在 CPU 1 上试图拿锁，开始自旋。
此时，虽然 CPU 0 可能还会运行其他任务，但任务 A 处于不可中断的睡眠状态（TASK_UNINTERRUPTIBLE），它必须等待某个特定事件（如硬件中断）来唤醒它。然而，自旋锁的临界区通常会**禁用本地中断（local_irq_disable）**或禁用抢占。如果唤醒 A 的中断恰好路由到了 CPU 1，而 CPU 1 正被任务 B 占着并处于自旋忙等状态（且可能也禁用了中断），那么中断无法被及时处理，A 就永远等不到唤醒信号。最终，CPU 1 的看门狗（Watchdog）会检测到 CPU 1 长时间“软锁死”（Soft Lockup），系统报错并可能崩溃。

简单来说：自旋锁的“自旋”要求持有者必须尽快运行并释放锁，而“睡眠”直接剥夺了持有者运行的机会，导致等待者永远等不到结果。

同步机制

完成量

Linux内核中的完成量是一种轻量级的同步机制，主要是用于让一个执行单元等待另一个执行单元完成某个特定的任务之后，再继续执行

数据结构定义如下：

struct completion {
    unsigned int done;
    struct swait_queue_head wait;
};

done: 一个无符号整数计数器。

done > 0：表示“已完成”事件已发生，等待者可立即继续。
done == 0：表示事件未发生，等待者需进入睡眠

wait: 一个等待队列头。当 done == 0 时，所有等待该完成量的进程都会在此队列中睡眠

初始化完成量

init_completion(struct completion *x)

等待完成量

wait_for_completion(struct completion *x);

在函数内部主要是获取得到这个完成量的等待队列节点，然后将这个等待队列节点上锁，在里面去使用一个do while循环等待，循环内部使用了调度回调函数action

这里使用一个do while循环主要是因为可能一次调度结束回来之后done还是为0,所以还需要继续去等待，这里可以看到等待的时候设置的状态是TASK_UNINTERRUPTIBLE不可中断状态

该状态只能使用wake_up函数唤醒

void __sched wait_for_completion(struct completion *x)
{
    wait_for_common(x, MAX_SCHEDULE_TIMEOUT, TASK_UNINTERRUPTIBLE);
}
//-------------------------------------
static inline long __sched
do_wait_for_common(struct completion *x,
           long (*action)(long), long timeout, int state)
{
    if (!x->done) {
        DECLARE_SWAITQUEUE(wait);

        do {
            if (signal_pending_state(state, current)) {
                timeout = -ERESTARTSYS;
                break;
            }
            __prepare_to_swait(&x->wait, &wait);
            __set_current_state(state);
            raw_spin_unlock_irq(&x->wait.lock);
            timeout = action(timeout);
            raw_spin_lock_irq(&x->wait.lock);
        } while (!x->done && timeout);
        __finish_swait(&x->wait, &wait);
        if (!x->done)
            return timeout;
    }
    if (x->done != UINT_MAX)
        x->done--;
    return timeout ?: 1;
}

static inline long __sched
__wait_for_common(struct completion *x,
          long (*action)(long), long timeout, int state)
{
    might_sleep();

    complete_acquire(x);

    raw_spin_lock_irq(&x->wait.lock);
    timeout = do_wait_for_common(x, action, timeout, state);
    raw_spin_unlock_irq(&x->wait.lock);

    complete_release(x);

    return timeout;
}

static long __sched
wait_for_common(struct completion *x, long timeout, int state)
{
    return __wait_for_common(x, schedule_timeout, timeout, state);
}

唤醒完成量

//唤醒等待队列的第一个节点
complete(struct completion *x);
//唤醒等待队列的所有节点
complete_all(struct completion *x);

函数的实现如下：

void complete(struct completion *x)
{
    unsigned long flags;

    raw_spin_lock_irqsave(&x->wait.lock, flags);

    if (x->done != UINT_MAX)
        x->done++;
    swake_up_locked(&x->wait);
    raw_spin_unlock_irqrestore(&x->wait.lock, flags);
}

这里首先将done值++，然后旧调用了swake_up_locked去唤醒等待队列的第一个节点

void swake_up_locked(struct swait_queue_head *q)
{
    struct swait_queue *curr;

    if (list_empty(&q->task_list))
        return;

    curr = list_first_entry(&q->task_list, typeof(*curr), task_list);
    wake_up_process(curr->task);
    list_del_init(&curr->task_list);
}

而complete_all就是循环调用wake_up_process去唤醒所有的节点

信号量

定义信号量

struct semaphore mySem;

初始化信号量

static inline void sema_init(struct semaphore *sem, int val)

获取信号量

void down(struct semaphore *sem)  //阻塞
void down_interruptible(struct semaphore *sem);  //非阻塞，尝试获取

释放信号量

void up(struct semaphore *sem)

一	二	三	四	五	六	日
	1	2	3	4	5	6
7	8	9	10	11	12	13
14	15	16	17	18	19	20
21	22	23	24	25	26	27
28	29	30