Linux内核深入学习 - 内核同步

目录

内核抢占

同步原语

per-CPU变量

API

Per CPU 变量的应用

per CPU 变量在多文件下的用法

原子操作

API

优化和内存屏障

自旋锁

自旋锁 API 函数

读写锁

API

RCU

信号量

API

1. 信号量的结构：

2. 初始化函数sema_init

3. 可中断获取信号量函数down_interruptible

4. 释放信号量函数 up

Reference

 在我们开始之前，先来看看内核抢占，他是我们为了解决的东西而引出来的内核同步！

内核抢占

在Linux中内核抢占比较复杂：

• 无论在抢占内核还是在非抢占内核，运行在内核态的程序都可以自动放弃CPU，比如说：其原因可能是竞争，由于等待资源，而不得不进入睡眠状态。往往把这种状态称为计划性进程切换，但是抢占式内核在响应引起进程切换的异步事件，例如说唤醒高优先级进程的中断处理程序的方式上，与非抢占的内核是有差别的。我们把这种进程切换称之为强制性的进程切换

• 所有的进程切换都由宏：switch_to来完成，在抢占式的内核和非抢占式的内核中，当进程的执行完某一些具有内核功能的线程，而且调度程序被调度后，就会发生进程切换！不过在非抢占内核中，当前进程是不可能被替换！除非他打算换到用户态上去！

因此抢占式内核的主要特点：是一个在内核态运行的进程可能在执行内核函数期间被另一个进程取代！

在Linux中当被current thread info宏所引用的thread_info描述符的preempt_count字段大于零时，就会禁止内核抢占他！

在如下任何一种情况发生时，取值都大于零：

1. 内核正在执行内中断服务例程

2. 可延迟函数被禁止

3. 通过把抢占计数器设置为正数而显示的禁用内核抢占

上面的原则告诉我们只有当内核正在执行异常处理程序，而且内核抢占没有被显式地禁用的时候，才会抢占内核

那么什么时候同步是必须的呢：我们之前就有提到过刚计算的结果依赖于两个或两个以上的交叉内核控制路径的嵌套方式时，才有可能会引起竞争！（说白了就是两个执行流撞在一起，有多个进程同时执行同一段代码），临界区是一段代码，在其他的内核控制路径能够进入临界区前，进入临界区的内核控制路径必须全部执行完！这段临界区的代码交叉内核控制路径使内核开发的工作者变得复杂，他们必须小心地识别出异常处理程序，中断处理程序，可延迟函数和内核进程中的临界区。一旦临界区被确定，就必须对其采取一定的保护措施

那么什么时候同步是不必要的呢：

• 所有中断处理程序响应来自pic的中断并且禁用了IRQ线，此外在中断处理程序的结束之前不允许产生相同的中断事件！

• 中断处理程序，软中断，tasklet既不可以被抢占也不可能被阻塞，所以他们不可能长时间的处于挂起状态！即使在最坏的情况下它们的执行也只是有轻微的延迟，因为在其执行的过程中可能会发生其他中断执行！

• 中断处理的内核控制路径不能被执行可延迟函数，或系统调用服务例程的内核控制路径所中断

• 软中断和tasklet不能在一个给定的CPU上交错执行

• 同一个tasklet不可能同时在几个CPU上执行！

以上的每一种设计选择都可以看作是一种约束，下面是一些可能简化了的例子

• 已中断处理程序和tasklet不必编写成可重入的函数

• 仅被软中断和task light访问的每CPU变量并不需要同步

• 仅被一种tasklet访问的数据结构是不需要同步的

同步原语

下面来看看同步原语，内核中使用的同步技术

per-CPU变量

最好的同步技术就是把不需要同步的内核放在首位正如我们将要看到事实上每一种显示的同步原语都会有不容忽视的性能开销最简单也是最重要的同步技术包括把内核变量声明为每CPU变量每CPU变量主要是数据结构的数组系统的每个CPU都对应数组的一个元素一个CPU不应该访问其他CPU对应的数组元素另外它可以随意读或修改他们自己的元素而不必担心竞争条件因为这是他唯一有资格这么做的CPU但是这也意味着每CPU变量基本上只能在特殊情况下才能够使用也就是当他确定这个系统上的CPU上的数据的逻辑上是独立的此外在单处理器和多处理器系统中内核抢占都可能会使每cpu变量产生竞争条件总的原则是内核控制路径应该禁用抢占的情况下去访问每cpu变量

API

为每个CPU定义一个变量的拷贝的宏定义在文件include/linux/percpu-defs.h中，如下：

#define DEFINE_PER_CPU(type, name) \
        DEFINE_PER_CPU_SECTION(type, name, "")

若我们使用DEFINE_PER_CPU(int, per_cpu_n)为每个CPU定义变量，其展开宏如下。

__attribute__((section(".data..percpu"))) int per_cpu_n
​
#define DEFINE_PER_CPU_SECTION(type, name, sec) \
    __PCPU_ATTRS(sec) __typeof__(type) name
#define __PCPU_ATTRS(sec)                        \
    __percpu __attribute__((section(PER_CPU_BASE_SECTION sec)))    
​
#define PER_CPU_BASE_SECTION ".data..percpu"

在链接过程中，所有通过DEFINE_PER_CPU宏定义的变量都将链接到一起。在操作系统启动时，Linux 将为该段分配一段内存。 查看编译出的内核镜像可找到.data..percpu段

# readelf -S vmlinux
  [Nr] Name              Type             Address           Offset
       Size              EntSize          Flags  Link  Info  Align
  [21] .data..percpu     PROGBITS         0000000000000000  01000000
       000000000001d000  0000000000000000  WA       0     0     4096

per CPU 变量的访问通过宏get_cpu_var完成 。Linux内核是可抢占的，并且在访问访问 per cpu变量时我们需要知道当前代码运行在哪个CPU核上。 因此，在访问每个cpu变量时，应当不允许抢占当前代码并将其移至另一个CPU。例如，若在获取到 CPU id 为 1 后，该任务被抢占而移动到了 CPU 2上继续运行，这时访问的将仍然是 CPU 1的per cpu 变量。因此，在 get_cpu_var 宏中，首先要调用preempt_disable()函数禁止任务抢占。

// in  include/linux/percpu-defs.h
#define get_cpu_var(var)                        \
(*({                                    \
    preempt_disable();                        \
    this_cpu_ptr(&var);                        \
}))

我比较好奇的是 this_cpu_ptr是如何实现的。内核如何将该变量对应到属于该CPU的 per CPU 变量内存呢？

在初始化时，内核会使用一个数组__per_cpu_offset[cpu]记录每个CPU静态per cpu 变量的偏移地址。在ARM64架构下， OS 启动时将 per cpu 偏移地址写入到 TPDIR_EL1 和 TPDIR_EL2 寄存器中。

void __init setup_per_cpu_areas(void)
{
    unsigned long delta;
    unsigned int cpu;
    ...
    delta = (unsigned long)pcpu_base_addr - (unsigned long)__per_cpu_start;
    for_each_possible_cpu(cpu)
        __per_cpu_offset[cpu] = delta + pcpu_unit_offsets[cpu];
}
​
/* arch/arm64/include/asm/percpu.h */
static inline void set_my_cpu_offset(unsigned long off)
{
    asm volatile(ALTERNATIVE("msr tpidr_el1, %0",
                 "msr tpidr_el2, %0",
                 ARM64_HAS_VIRT_HOST_EXTN)
            :: "r" (off) : "memory");
}

this_cpu_ptr的宏展开如下：即相当于 percpu 变量指针 ptr 加上__my_cpu_offset。

#define arch_raw_cpu_ptr(ptr) SHIFT_PERCPU_PTR(ptr, __my_cpu_offset)
#define raw_cpu_ptr(ptr)                        \
({                                    \
    __verify_pcpu_ptr(ptr);                        \
    arch_raw_cpu_ptr(ptr);                        \
})
​
#define this_cpu_ptr(ptr)    raw_cpu_ptr(ptr)

__my_cpu_offset宏即是从当前cpu的tpidr_el1、tpidr_el2寄存器中取出此前设置的__per_cpu_offset[cpu]值，实现如下：

static inline unsigned long __my_cpu_offset(void)
{
    unsigned long off;
​
    /*
     * We want to allow caching the value, so avoid using volatile and
     * instead use a fake stack read to hazard against barrier().
     */
    asm(ALTERNATIVE("mrs %0, tpidr_el1",
            "mrs %0, tpidr_el2",
            ARM64_HAS_VIRT_HOST_EXTN)
        : "=r" (off) :
        "Q" (*(const unsigned long *)current_stack_pointer));
​
    return off;
}

有时会需要指定某个 CPU 获取其某个 per cpu 变量的地址，通过宏per_cpu_ptr实现，源码如下：

#define SHIFT_PERCPU_PTR(__p, __offset)                    \
    RELOC_HIDE((typeof(*(__p)) __kernel __force *)(__p), (__offset))
​
#define per_cpu_ptr(ptr, cpu)                        \
({                                    \
    __verify_pcpu_ptr(ptr);                        \
    SHIFT_PERCPU_PTR((ptr), per_cpu_offset((cpu)));            \
})

user，kernel，safe，force等定义在compiler_type.h头文件中。看到两个很奇怪的现象，一个是只有在CHECKER宏打开的情况下，他们的定义才会被实现，否则他们的定义是空的。第二个是它们的attribute的定义，并不是gcc支持的属性。那到底是哪里使用到了呢？原来linux的作者们自己开发了一套编译期代码检查的工具Sparse，可以用于在编译阶段快速发现代码中隐含的问题。[1]

• address_space 定义了指针能指向的内存的类型，0代表kernel space，1代表user space，2代表设备地址空间，3代表cpu局部的内存空间

• safe 表示变量可以为空

• force 表示变量可以强制类型转换

Per CPU 变量的应用

记录每个CPU 的 id 是 per CPU 变量的应用之一。

那么有了 Per CPU变量之后，如何获得当前执行代码的CPU 编号？ 内核函数smp_processor_id()用来获取当前 CPU 的 id 。

CPU id 的存储依赖于 per CPU 变量（DEFINE_PER_CPU宏用来定义 cpu_number 变量）。

// 每个CPU的cpuid是放置在cpu_number这个percpu变量中
DEFINE_PER_CPU(int, cpu_number);

在内核初始化时，smp_prepare_cpus()函数执行per_cpu(cpu_number, cpu) = cpu;设定每个核的编号。

// in /arch/arm64/kernel/smp.c
void __init smp_prepare_cpus(unsigned int max_cpus)
{
    const struct cpu_operations *ops;
    int err;
    unsigned int cpu;
    unsigned int this_cpu;
​
    init_cpu_topology();
​
    this_cpu = smp_processor_id();
​
    for_each_possible_cpu(cpu) {
        // 设置 CPU id
        per_cpu(cpu_number, cpu) = cpu;
        // 确定在哪个核上执行的，若是本身则跳过。
        if (cpu == smp_processor_id())
            continue;
​
        ops = get_cpu_ops(cpu);
        if (!ops)
            continue;
​
        err = ops->cpu_prepare(cpu);
        if (err)
            continue;
​
        set_cpu_present(cpu, true);
        numa_store_cpu_info(cpu);
    }
}

smp_processer_id ()函数(定义在 include/linux/smp.h)展开如下。

# define smp_processor_id() __smp_processor_id()
#define __smp_processor_id(x) raw_smp_processor_id(x)

raw_smp_processor_id与处理器架构相关（下例为ARM64）的实现如下，raw_cpu_ptr 获取到 cpu_number 的地址，在解引用得到 cpu id。

#define raw_smp_processor_id() (*raw_cpu_ptr(&cpu_number))

per CPU 变量在多文件下的用法

声明一个 per cpu 变量并在另一个文件中引用，以获取当前 task_struct 为例（x86下 current 宏的实现）。

定义方式如下：

DEFINE_PER_CPU(struct task_struct *, current_task) ____cacheline_aligned =
    &init_task;
EXPORT_PER_CPU_SYMBOL(current_task);

在另一个文件中引用方式如下：

DECLARE_PER_CPU(struct task_struct *, current_task);
static __always_inline struct task_struct *get_current(void)
{
    return this_cpu_read_stable(current_task);
}
​
#define current get_current()

原子操作

若干汇编语言指令是具有RCU类型的，也就是说他们访问存储器单元两次！第一次读原值，第二次写新值

避免由于RCU指令引起的竞争条件的容易的办法，就是确保这样的操作在芯片级就是原子性的！

任何一个这样的操作都必须单个指令进行执行中间，是不允许中断的且避免其他的CPU访问统一存储器单元。这些很小的原子操作可以建立在其他更灵活的机制的基础之上创建临界区

让我们根据这样的分类来回顾一下8086的指令：

进行零次或一次对齐内存访问的汇编指令是原子的

如果在读操作之后写操作之前没有其他处理器占用内存总线，那么在从内存中读取数据，更新数据并且把更新后的数据写回内存中的这些RCU汇编语言指令是原子的！当然在单处理器系统中永远不会发生内存总线窃取的情况

操作码前缀是lock字节(0xf0)的汇编语言指令，即使在多处理器系统中，也是原子的！当控制单元检测到这个前缀时，就会锁定内存总线直到这条指令完成为止，因此在加速的指令执行时其他处理器是不能够访问这个内存单元的！

操作码前缀是reg字节（0xf2, 0xf3）的汇编语言指令不是原子的！这条指令强行让控制单元多次重复相同的指令控制单元，在执行新的循环之前要检查挂起的中断！

API

相应的也提供了 64 位原子变量的操作 API 函数，这里我们就不详细讲解了，和表 中的 API 函数有用法一样，只是将“atomic”前缀换为“atomic64”，将 int 换为 long long。如果使用的是 64 位的 SOC，那么就要使用 64 位的原子操作函数。

优化和内存屏障

当使用边缘优化的编译器时，它会重排汇编指令从而达到以最优，此外现代CPU通常会并行地执行若干条指令且可能重新安排内存访问这种重新排序可以极大地加速程序的执行然而当处理同步时则必须避免指令重新排序优化屏障源于保证编译程序不会混淆放在原语操作之前的汇编语言指令和放在原语操作之后的汇编语言这些汇编语言指令在C中都有对应的语句在Linux中优化屏障就是barrier宏，它展开为：

asm volitile("":::"memory");

指令ASM告诉编译程序要插入汇编语言片段，volatile关键字禁止编译器把ASM指令与程序中的其他指令重新组合！memory关键字强制编译器假定RAM中的所有内存单元已经被汇编语言指令修改，因此编译器不能使用存放在CPU寄存器中的内存单元的值来优化ASM指令前的代码

注意优化屏障并不保证不使当前CPU把汇编语言指令混在一起执行，这是内存屏障的工作！

内存屏障源于保证在原语之后的操作开始之前，原语之前的操作已经完成。因此内存屏障类似于防火墙，让任何汇编语句指令都不能通过！在以下这些汇编指令在8086处理中是串行的！因为他们起到了内存屏障的作用：

对IO端口进行操作的所有指令

有lock前缀的所有指令

写控制寄存器系统寄存器或调试寄存器的所有指令

在奔腾4微处理器中引入的汇编指令lfence, sfence, mfence

关于内存屏障的汇编指令少数专门的汇编语言指令

Linux使用六个内存屏障原语，如下表所示：

自旋锁

这是一种广泛使用的锁，关于锁，可以认为是对访问公共资源的一种限制。如果内核控制路径希望访问资源，就必须获取钥匙来打开这个锁！当且只当资源空闲时，也就是没有任何进程来访问这段资源的时候，它才能成功，然后持有这个锁！其他进程想要在这个进程处理。这个数据结构的时候必须等待这个进程处理完毕，释放掉这个锁之后，其他进程才能够接着访问这个数据结构！

严肃的版本：

自旋锁是用来在多处理器环境中工作的一种特殊的锁，如果内核控制路径发现自旋锁是开着的！那么获取锁，并且继续执行，相反，则会在周围旋转反复执行一条紧凑的循环指令，直到锁被释放！

自旋锁的循环指令表示忙等待，即使等待的内核控制路径是无事可做的。它也会在CPU上保持运行，不过自旋锁通常非常方便。因为很多内核资源只锁一毫秒的时间片段，所以说释放CPU和随后又获得CPU是不会消耗多少时间的！

这里有更加详细的自旋锁的文章，可以参看：

[Linux中的spinlock机制一] - CAS和ticket spinlock - 知乎 (zhihu.com)]

自旋锁 API 函数

最基本的自旋锁 API 函数如下表 所示：

读写锁

读写自旋锁的引入是为了增加内核的并发能力，因为我们思考：只要没有内核控制路径对希望上锁的数据结构进行修改，我们就没有必要对这个数据结构进行上锁。只有当我们想要对这个结构进行写操作的时候那么我们才会对这个资源进行上锁。

API

与 spinlock 一样，Read/Write spinlock 有如下的 APIs：

RCU

RCU就是Read, Copy, Update机制，这是为了保护在多数情况下被多个CPU读的数据结构，而设计的一种同步技术！它允许多个读者和写者并发执行，而且它是不会使用锁的。就是说它不使用被所有CPU共享的锁或计数器

在这一点上与读写自旋锁与顺序锁相比，它具有更大的优势！它的关键思想在于限制RCP的范围：

• RCU只会保护被动态分配，并通过指针引用的数据结构

• 在被RCU保护的临界区中任何内核控制路径都不能睡眠

信号量

它本质上就是一个更加高级的锁，一个允许最大若干进程访问资源的锁。

Linux有两个信号量：内核信号量和IPC信号量，我们现在只关心前者

API

注意：down_interruptible 接口，在获取不到信号量的时候，该任务会进入 INTERRUPTABLE 的睡眠，但是 down() 接口会导致进入 UNINTERRUPTABLE 的睡眠，down 用的较少。

1. 信号量的结构：

struct semaphore {
    raw_spinlock_t      lock;
    unsigned int        count;
    struct list_head    wait_list;
};

信号量用结构semaphore描述，它在自旋锁的基础上改进而成，它包括一个自旋锁、信号量计数器和一个等待队列。用户程序只能调用信号量API函数，而不能直接访问信号量结构。

2. 初始化函数sema_init

#define __SEMAPHORE_INITIALIZER(name, n)                \
{                                   \
    .lock       = __RAW_SPIN_LOCK_UNLOCKED((name).lock),    \
    .count      = n,                        \
    .wait_list  = LIST_HEAD_INIT((name).wait_list),     \
}
 
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);
}

初始化了信号量中的 spinlock 结构，count 计数器和初始化链表。

3. 可中断获取信号量函数down_interruptible

static noinline int __sched __down_interruptible(struct semaphore *sem)
{
    return __down_common(sem, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}
 
int down_interruptible(struct semaphore *sem)
{
    unsigned long flags;
    int result = 0;
 
    raw_spin_lock_irqsave(&sem->lock, flags);
    if (likely(sem->count > 0))
        sem->count--;
    else
        result = __down_interruptible(sem);
    raw_spin_unlock_irqrestore(&sem->lock, flags);
 
    return result;
}

down_interruptible 进入后，获取信号量获取成功，进入临界区，否则进入 down_interruptible->down_common

static inline int __sched __down_common(struct semaphore *sem, long state,
                                long timeout)
{
    struct semaphore_waiter waiter;
 
    list_add_tail(&waiter.list, &sem->wait_list);
    waiter.task = current;
    waiter.up = false;
 
    for (;;) {
        if (signal_pending_state(state, current))
            goto interrupted;
        if (unlikely(timeout <= 0))
            goto timed_out;
        __set_current_state(state);
        raw_spin_unlock_irq(&sem->lock);
        timeout = schedule_timeout(timeout);
        raw_spin_lock_irq(&sem->lock);
        if (waiter.up)
            return 0;
    }
 
 timed_out:
    list_del(&waiter.list);
    return -ETIME;
 
 interrupted:
    list_del(&waiter.list);
    return -EINTR;
}

加入到等待队列，将状态设置成为 TASK_INTERRUPTIBLE ， 并设置了调度的 Timeout ： MAX_SCHEDULE_TIMEOUT

在调用了 schedule_timeout，使得进程进入了睡眠状态。

4. 释放信号量函数 up

void up(struct semaphore *sem)
{
    unsigned long flags;
 
    raw_spin_lock_irqsave(&sem->lock, flags);
    if (likely(list_empty(&sem->wait_list)))
        sem->count++;
    else
        __up(sem);
    raw_spin_unlock_irqrestore(&sem->lock, flags);
}

如果等待队列为空，没有睡眠的进程期望获取这个信号量，则直接 count++，否则调用 __up：

static noinline void __sched __up(struct semaphore *sem)
{
    struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
                        struct semaphore_waiter, list);
    list_del(&waiter->list);
    waiter->up = true;
    wake_up_process(waiter->task);
}static noinline void __sched __up(struct semaphore *sem)
{
    struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
                        struct semaphore_waiter, list);
    list_del(&waiter->list);
    waiter->up = true;
    wake_up_process(waiter->task);
}

取出队列中的元素，进行唤醒操作。

Reference

Linux--原子操作（介绍及其操作函数集）_原子操作函数-CSDN博客

一文读懂优化屏障和内存屏障 - 知乎 (zhihu.com)

[Linux中的RCU机制一] - 原理与使用方法 - 知乎 (zhihu.com)

Linux 内核同步（五）：信号量（semaphore）_sema_init-CSDN博客