松垮垮
songkuakua
2025-02-15
目录

Linux内核同步机制原理与实现

# Linux内核同步机制原理与实现

Owner: -QVQ-

一文搞懂Linux内核同步机制原理与实现(上篇) (opens new window)

一文搞懂Linux内核同步机制原理与实现(下篇) (opens new window)

# 中断概念

**上半部：**指的是中断处理程序，适用一个任务对时间非常敏感，一个任务和硬件相关 ，一个任务要保证不被其他中断（特别是相同的中断）打断

**下半部：**则指的是一些虽然与中断有相关性但是可以延后执行的任务。除了上面提到上半部需要的情况，其他都可以放在下半部

中断不能被相同类型的中断打断，而下半部依然可以被中断打断；

**软中断：**作为下半部机制的代表，是随着SMP（share memory processor）的出现应运而生的，它也是tasklet实现的基础。它能实现下半部的功能，使得对时间不敏感的任务延后执行，而且可以在多个CPU上并行执行

产生后要等待内核的调度才能执行。软中断不能被自己打断(即单个cpu上软中断不能嵌套执行)，只能被硬件中断打断（上半部）

可以并发运行在多个CPU上（即使同一类型的也可以）。所以软中断必须设计为可重入的函数（允许多个CPU同时操作），因此也需要使用自旋锁来保其数据结构。

**tasklet：**软中断的一种特殊用法。如果某种延迟应用并不需要在多个CPU上并行执行，软中断显得设计复杂度变高，增加开发负担。taklet的意义在这里

一种特定类型的tasklet只能运行在一个CPU上，只能串行执行。

多个不同类型的tasklet可以并行在多个CPU上。

软中断是静态分配的，在内核编译好之后，就不能改变。但tasklet就灵活许多，可以在运行时改变（比如添加模块时）。

# 中断API

#include <asm/irqflags.h>

关中断：local_irq_save(flags)//flags是unsigned long,当前中断允许状态保存在flags中 开中断：local_irq_restore(flags) 关中断：local_irq_disable() 开中断：local_irq_enable()

# 原子操作

原子操作主要用于实现资源计数

原子操作与硬件架构强相关，其API具体的定义均位于对应arch目录下的include/asm/atomic.h文件中，通过汇编语言实现，内核源码根目录下的include/asm-generic/atomic.h则抽象封装了API，该API最后分派的实现来自于arch目录下对应的代码。

# 数据结构如下

typedef struct{
	int counter;
}atomic_t;
1
2
3

# API

int atomic_read(atomic_t**v)读操作*

*void atomic_set(atomic_t***v,int i)设置变量*

void atomic_add(int i,atomic_t*v)增加i

void atomic_sub(int i,atomic_t*v)减少ⅰ

void atomic_inc(atomic_t*v)增加1

void atomic_dec(atomic_t*v)减少1 bool atomic_inc_and_test(atomic_t*v)加1是否为0

bool atomic_dec_and_test(atomic_t*v)减1是否为0

bool atomic_add_negative(int i,atomic_t*v)加i是否为负

int atomic_add_return(int i,atomic_t *v)增加ⅰ返回结果

int atomic_sub_return(int i,atomic_t *v)减少ⅰ返回结果

int atomic_inc_return(int i,atomic_t *v)加1返回

int atomic_dec_return(int i,atomic_t *v)减1，返回

# 自旋锁（spinlock)

如果只有进程上下文的访问，那么可以考虑使用semaphore或者mutex的锁机制，但如果中断上下文也要访问这个资源，则可能导致睡眠的lock不能用。（中断上下文，是不允许睡眠）

中断处理的时候,不应该发生进程切换：

  1. 因为在中断context中，唯一能打断当前中断handler的只有更高优先级的中断。如果在中断context中休眠，则没有办法唤醒它
  2. schedule()在切换进程时，保存当前的进程上下文，但在中断处理程序里，CPU寄存器的值肯定已经变化了
  1. 内核中schedule()函数本身在进来的时候判断是否处于中断上下文:

if(unlikely(in_interrupt()))

BUG();

因此，强行调用schedule()的结果就是内核BUG。

若自旋锁已被别的执行者保持，调用者就会原地循环等待并检查该锁的持有者是否已经释放锁（即进入自旋状态），若释放则调用者开始持有该锁。自旋锁持有期间不可被抢占。（自旋锁消耗CPU资源不应滥用）

# 使用场景

# 内核抢占场景

进程A和进程B访问共享资源R，如果A和B在同一CPU下，A上锁后进入临界区，发生中断，优先级更高的进程B想进入临界区而请求锁而陷入锁死（自旋锁的策略是一直循环不放弃CPU，等待资源）。

因此spinlock在访问临界区时关掉中断。在多CPU下虽然不会死锁但会降低效率

# 中断上下文场景

进程A、B和外设P，CPU0、1，资源R

进程A拿到lock访问R，外设P的中断事件调度到CPU0访问R，循环等待lock而死锁，CPU1此时也访问R也陷入死锁

因此spinlock在访问临界区时，如果涉及到中断上下文的访问，spin lock需要和禁止本 CPU 上的中断联合使用。

# 底半部场景

……

# 数据结构如下

typedef struct {
	union {
		struct raw_spinlock rlock;
	}
}spinlock_t;

typedef struct{
	arch_spinlock_t raw_lock;
}	raw_spinlock_t;

typedef struct{
	union{
		u32 slock;//spinlock根本的实现依赖于slock这个变量
		struct __raw_tickets{
#ifdef ARMEB
	//根据字节顺序不同，安排变量内存布局
	u16 next;ul6 owner;
#else
	ul6 owner;ul6 next;
#endif
		};
	};
}arch_spinlock_t;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
//动态定义一个自旋锁
spinlock_t lock;
spin_lock_init (&lock);
//静态定义一个自旋锁
DEFINE_SPINLOCK(lock);
1
2
3
4
5

# API

  • static inline void arch_spin_unlock(arch_spinlock_t *lock)

    static inline void arch_spin_unlock(arch_spinlock_t *lock){
	smp_mb（）;
	lock->tickets.owner++;
	dsb_sev（）;
}
    1
    2
    3
    4
    5

  • static inline void arch_spin_lock(arch_spinlock_t *lock)

    static inline void arch_spin_lock(arch_spinlock_t *lock)
{
	unsigned long tmp;
	u32 newval;
	arch_spinlock_t lockval;

	prefetchw(&lock->slock);
	__asm__ __volatile__(
"1:	ldrex	%0, [%3]\n"
"	add	%1, %0, %4\n"
"	strex	%2, %1, [%3]\n"
"	teq	%2, #0\n"
"	bne	1b"
	: "=&r" (lockval), "=&r" (newval), "=&r" (tmp)
	: "r" (&lock->slock), "I" (1 << TICKET_SHIFT)
	: "cc");

	while (lockval.tickets.next != lockval.tickets.owner) {
		wfe();
		lockval.tickets.owner = READ_ONCE(lock->tickets.owner);
	}

	smp_mb();
}
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24

在lock中prefetchw预加载slock的值，slock和next、owner联合共享空间，在unlock函数中owner+1，lock函数中比较owner和next中的值，不一致则一直循环，相等则结束等待。

# 其他自旋锁的变体API

接口API的类型 spinlock中的定义 raw_spinlock的定义
定义spin lock并初始化 DEFINE_SPINLOCK DEFINE_RAW_SPINLOCK
动态初始化spin lock spin_lock_init raw_spin_lock_init
获取指定的spin lock spin_lock raw_spin_lock
获取指定的spin lock同时disable本CPU中断 spin_lock_irq raw_spin_lock_irq
保存本CPU当前的irq状态，disable本CPU中断并获取指定的spin lock spin_lock_irqsave raw_spin_lock_irqsave
获取指定的spin lock同时disable本CPU的bottom half spin_lock_bh raw_spin_lock_bh
释放指定的spin lock spin_unlock raw_spin_unlock
释放指定的spin lock同时enable本CPU中断 spin_unlock_irq raw_spin_unock_irq
释放指定的spin lock同时恢复本CPU的中断状态 spin_unlock_irqstore raw_spin_unlock_irqstore
获取指定的spin lock同时enable本CPU的bottom half spin_unlock_bh raw_spin_unlock_bh
尝试去获取spin lock，如果失败，不会spin，而是返回非零值 spin_trylock raw_spin_trylock
判断spin lock是否是locked，如果其他的thread已经获取了该lock，那么返回非零值，否则返回0 spin_is_locked raw_spin_is_locked

# 不同版本spin_lock的使用

性能上，spin_lock > spin_lock_bh > spin_lock_irq > spin_lock_irqsave。

安全上，spin_lock_irqsave > spin_lock_irq > spin_lock_bh >spin_lock。

spin_lock用于阻止在不同CPU上的执行单元对共享资源的同时访问以及不同进程上下文互相抢占导致的对共享资源的非同步访问，

而中断失效（spin_lock_irq）和软中断失效(spin_lock_bh)却是为了阻止在同一CPU上软中断或中断对共享资源的非同步访问。

如果被保护的共享资源在多个tasklet或timer上下文访问，那么对共享资源的访问仅需要用spin_lock和spin_unlock来保护。（只需要解决不同CPU的并行问题就好，不用解决同一CPU的软中断）

如果被保护的共享资源只在一个软中断（tasklet和timer除外）上下文访问，那么这个共享资源需要用spin_lock和spin_unlock来保护，因为同样的软中断可以同时在不同的CPU上运行。

如果被保护的共享资源只在进程上下文访问和多个软中断上下文（包含tasklet或timer）访问，对共享资源的访问最好使用spin_lock_bh和spin_unlock_bh来保护。（因为当在进程上下文访问共享资源时，可能被软中断打断，从而可能进入软中断上下文来对被保护的共享资源访问）

如果被保护的共享资源在软中断（包括tasklet和timer）或进程上下文和硬中断上下文访问，在进程或软中断上下文需要使用spin_lock_irq和spin_unlock_irq来保护对共享资源的访问。那么在软中断或进程上下文访问期间，可能被硬中断打断，从而进入硬中断上下文对共享资源进行访问。

在使用spin_lock_irq和spin_unlock_irq的情况下，完全可以用spin_lock_irqsave和spin_unlock_irqrestore取代

# 信号量（semaphore）

# 读写信号量（rw_semaphore）

# Spinlock

# Mutex

# BKL(Big Kernel Lock，只包含在2.4内核中，不讲)

# Rwlock

# RCU机制

RCU（Read-Copy Update）是数据同步的一种方式，主要针对的数据对象是链表。这是 Linux 内核实现的一种针对“读多写少”的共享数据的同步机制。起到替代加锁操作的效果。

这样在同一时间可以有多个线程同时读取该链表，并且允许一个线程对链表进行修改（修改的时候，需要加锁）。RCU适用于需要频繁的读取数据，而相应修改数据并不多的情景。

延后的删除或释放将占用一些内存，这是它的缺点

# 实现机制：

1、在读取过程中，另外一个线程删除了一个节点。删除线程可以把这个节点从链表中移除，但它不能直接销毁这个节点，必须等到所有的读取线程读取完成以后，才进行销毁操作。RCU中把这个过程称为宽限期（Grace period）。

2、在读取过程中，另外一个线程插入了一个新节点，而读线程读到了这个节点，那么需要保证读到的这个节点是完整的。这里涉及到了发布-订阅机制（Publish-Subscribe Mechanism）。

3、写者在访问它时首先拷贝一个副本，然后对副本进行修改，最后使用一个回调（callback）机制在适当的时机把指向原来数据的指针重新指向新的被修改的数据。

# API

保持一个读者的RCU临界区.在该临界区内不允许发生上下文切换。

rcu_read_lock()

rcu_read_unlock()

  • rcu_dereference()：读者调用它来获得一个被RCU保护的指针。
  • rcu_assign_pointer()：写者使用该函数来为被RCU保护的指针分配一个新的值。

# 案例

# 读链表操作

仅仅是换了锁的函数

//原代码
static enum audit_state audit_filter_task(struct task_struct *tsk)
{
    struct audit_entry *e;
    enum audit_state  state;

    read_lock(&auditsc_lock);

    /* Note: audit_netlink_sem held by caller. */
    list_for_each_entry(e, &audit_tsklist, list) {

        if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {
            read_unlock(&auditsc_lock);
            return state;
        }
    }

    read_unlock(&auditsc_lock);

    return AUDIT_BUILD_CONTEXT;
}
///RCU代码
static enum audit_state audit_filter_task(struct task_struct *tsk)
{

    struct audit_entry *e;
    enum audit_state  state;

    rcu_read_lock();

    /* Note: audit_netlink_sem held by caller. */

    list_for_each_entry_rcu(e, &audit_tsklist, list) {
        if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {
            rcu_read_unlock();
            return state;
        }
    }

    rcu_read_unlock();
    return AUDIT_BUILD_CONTEXT;
}
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
31
32
33
34
35
36
37
38
39
40
41
42

# 删链表操作

对于链表删除操作，list_del替换为list_del_rcu和call_rcu

call_rcu起到延时删除的作用，待到所有进程读完成才会真正删除

static inline int audit_del_rule(struct audit_rule *rule,
                                 struct list_head *list)

{

    struct audit_entry *e;

    write_lock(&auditsc_lock);

    list_for_each_entry(e, list, list) {

        if (!audit_compare_rule(rule, &e->rule)) {
            list_del(&e->list);
            write_unlock(&auditsc_lock);
            return 0;
        }
    }

    write_unlock(&auditsc_lock);
    return -EFAULT;     /* No matching rule */
}

//CRU代码
static inline int audit_del_rule(struct audit_rule *rule,
                                 struct list_head *list)
{

    struct audit_entry *e;

    /* Do not use the _rcu iterator here, since this is the only
     * deletion routine. */

    list_for_each_entry(e, list, list) {
        if (!audit_compare_rule(rule, &e->rule)) {
            list_del_rcu(&e->list);
            call_rcu(&e->rcu, audit_free_rule, e);
            return 0;
        }
    }

    return -EFAULT;     /* No matching rule */
}
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
31
32
33
34
35
36
37
38
39
40
41
42
43

写需要较多操作，对于CRU，先要申请新的链表节点，将原有的值复制过来，再call_rcu，其中上锁修改指针的值

修改条目将被在经历一个grace period后安全删除。


static inline int audit_add_rule(struct audit_entry *entry,
                                 struct list_head *list)

{

    write_lock(&auditsc_lock);

    if (entry->rule.flags & AUDIT_PREPEND) {

        entry->rule.flags &= ~AUDIT_PREPEND;
        list_add(&entry->list, list);
    } else {
        list_add_tail(&entry->list, list);
    }

    write_unlock(&auditsc_lock);

    return 0;
}
//CRU

static inline int audit_upd_rule(struct audit_rule *rule,
                                 struct list_head *list,
                                 __u32 newaction,
                                 __u32 newfield_count)
{

    struct audit_entry *e;
    struct audit_newentry *ne;

    list_for_each_entry(e, list, list) {

        if (!audit_compare_rule(rule, &e->rule)) {

            ne = kmalloc(sizeof(*entry), GFP_ATOMIC);

            if (ne == NULL)
                return -ENOMEM;

            audit_copy_rule(&ne->rule, &e->rule);
            ne->rule.action = newaction;
            ne->rule.file_count = newfield_count;
            list_replace_rcu(e, ne);

            call_rcu(&e->rcu, audit_free_rule, e);

            return 0;
        }
    }

    return -EFAULT;     /* No matching rule */

}
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# 修改操作立即可见

如果读者不能够容忍修改在一段时间后看到，在每一个链表条目中增加了一个deleted字段，标记该字段是否删除，如果删除了，就设置为真

读端代码应该修改为：

static enum audit_state audit_filter_task(struct task_struct *tsk)
{
    struct audit_entry *e;
    enum audit_state  state;

    rcu_read_lock();

    list_for_each_entry_rcu(e, &audit_tsklist, list) {

        if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {
            spin_lock(&e->lock);
						//这里是需要加上的部分
            if (e->deleted) {
                spin_unlock(&e->lock);
                rcu_read_unlock();
                return AUDIT_BUILD_CONTEXT;
            }

            rcu_read_unlock();

            return state;
        }
    }

    rcu_read_unlock();

    return AUDIT_BUILD_CONTEXT;
}
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

写段加一行即可

e->deleted = 1;

# brlock（只包含在2.4内核中，不讲）

# seqlock（只包含在2.6内核及以后的版本中）

# 抢占

preempt_disable（）:关抢占

preempt_enable（）:开抢占

# 分析

一般关中断对系统的性能影响更大，所以关抢占

在一些场景里需要同时关中断和抢占，是为了避免复杂的开发中调用的函数里有打开调度的操作。

上次更新: 2025/02/21, 14:57:10
cpu页表
per

← per→

最近更新
01
搭建ai知识助手
02-23
02
边缘检测
02-15
03
css
02-15
更多文章>
Theme by Vdoing | Copyright © 2025-2025 松垮垮 | MIT License | 蜀ICP备2025120453号 | 川公网安备51011202000997号