极客时间已完结课程限时免费阅读

08 | 锁：并发操作中，解决数据同步的四种方法

08 | 锁：并发操作中，解决数据同步的四种方法-极客时间

08 | 锁：并发操作中，解决数据同步的四种方法

2021-05-26 LMOS 来自北京
《操作系统实战45讲》
课程介绍

讲述：陈晨

时长17:33大小16.04M

你好，我是 LMOS。
我们在前面的课程中探索了，开发操作系统要了解的最核心的硬件——CPU、MMU、Cache、内存，知道了它们的工作原理。在程序运行中，它们起到了至关重要的作用。
在开发我们自己的操作系统以前，还不能一开始就把机器跑起来，而是先要弄清楚数据同步的问题。如果不解决掉数据同步的问题，后面机器跑起来，就会出现很多不可预知的结果。
通过这节课，我会给你讲清楚为什么在并发操作里，很可能得不到预期的访问数据，还会带你分析这个问题的原因以及解决方法。有了这样一个研究、解决问题的过程，对最重要的几种锁（原子变量，关中断，信号量，自旋锁），你就能做到心中有数了。

非预期结果的全局变量

来看看下面的代码，描述的是一个线程中的函数和中断处理函数，它们分别对一个全局变量执行加 1 操作，代码如下。
int a = 0;
void interrupt_handle()
{
a++;
}
void thread_func()
{
a++;
}
首先我们梳理一下编译器的翻译过程，通常编译器会把 a++ 语句翻译成这 3 条指令。
1. 把 a 加载某个寄存器中。
2. 这个寄存器加 1。
3. 把这个寄存器写回内存。
那么不难推断，可能导致结果不确定的情况是这样的：thread_func 函数还没运行完第 2 条指令时，中断就来了。
因此，CPU 转而处理中断，也就是开始运行 interrupt_handle 函数，这个函数运行完 a=1，CPU 还会回去继续运行第 3 条指令，此时 a 依然是 1，这显然是错的。
下面来看一下表格，你就明白了。
显然在 t2 时刻发生了中断，导致了 t2 到 t4 运行了 interrupt_handle 函数，t5 时刻 thread_func 又恢复运行，导致 interrupt_handle 函数中 a 的操作丢失，因此出错。

方法一：原子操作 拿下单体变量

要解决上述场景中的问题，有这样两种思路。一种是把 a++ 变成原子操作，这里的原子是不可分隔的，也就是说要 a++ 这个操作不可分隔，即 a++ 要么不执行，要么一口气执行完；另一种就是控制中断，比如在执行 a++ 之前关掉中断，执行完了之后打开中断。
我们先来看看原子操作，显然靠编译器自动生成原子操作不太可能。第一，编译器没有这么智能，能检测哪个变量需要原子操作；第二，编译器必须要考虑代码的移植性，例如有些硬件平台支持原子操作的机器指令，有的硬件平台不支持原子操作。
既然实现原子操作无法依赖于具体编译器，那就需要我们自己动手，x86 平台支持很多原子指令，我们只需要直接应用这些指令，比如原子加、原子减，原子读写等，用汇编代码写出对应的原子操作函数就行了。
好在现代 C 语言已经支持嵌入汇编代码，可以在 C 函数中按照特定的方式嵌入汇编代码了，实现原子操作就更方便了，代码如下。
//定义一个原子类型
typedef struct s_ATOMIC{
volatile s32_t a_count; //在变量前加上volatile，是为了禁止编译器优化，使其每次都从内存中加载变量
}atomic_t;
//原子读
static inline s32_t atomic_read(const atomic_t *v)
{
//x86平台取地址处是原子
return (*(volatile u32_t*)&(v)->a_count);
}
//原子写
static inline void atomic_write(atomic_t *v, int i)
{
//x86平台把一个值写入一个地址处也是原子的
v->a_count = i;
}
//原子加上一个整数
static inline void atomic_add(int i, atomic_t *v)
{
__asm__ __volatile__("lock;" "addl %1,%0"
: "+m" (v->a_count)
: "ir" (i));
}
//原子减去一个整数
static inline void atomic_sub(int i, atomic_t *v)
{
__asm__ __volatile__("lock;" "subl %1,%0"
: "+m" (v->a_count)
: "ir" (i));
}
//原子加1
static inline void atomic_inc(atomic_t *v)
{
__asm__ __volatile__("lock;" "incl %0"
: "+m" (v->a_count));
}
//原子减1
static inline void atomic_dec(atomic_t *v)
{
__asm__ __volatile__("lock;" "decl %0"
: "+m" (v->a_count));
}
以上代码中，加上 lock 前缀的 addl、subl、incl、decl 指令都是原子操作，lock 前缀表示锁定总线。
我们还是来看看 GCC 支持嵌入汇编代码的模板，不同于其它 C 编译器支持嵌入汇编代码的方式，为了优化用户代码，GCC 设计了一种特有的嵌入方式，它规定了汇编代码嵌入的形式和嵌入汇编代码需要由哪几个部分组成，如下面代码所示。
__asm__ __volatile__(代码部分:输出部分列表: 输入部分列表:损坏部分列表);
可以看到代码模板从 __asm__ 开始（当然也可以是 asm），紧跟着 __volatile__，然后是跟着一对括号，最后以分号结束。括号里大致分为 4 个部分：
1. 汇编代码部分，这里是实际嵌入的汇编代码。
2. 输出列表部分，让 GCC 能够处理 C 语言左值表达式与汇编代码的结合。
3. 输入列表部分，也是让 GCC 能够处理 C 语言表达式、变量、常量，让它们能够输入到汇编代码中去。
4. 损坏列表部分，告诉 GCC 汇编代码中用到了哪些寄存器，以便 GCC 在汇编代码运行前，生成保存它们的代码，并且在生成的汇编代码运行后，恢复它们（寄存器）的代码。
它们之间用冒号隔开，如果只有汇编代码部分，后面的冒号可以省略。但是有输入列表部分而没有输出列表部分的时候，输出列表部分的冒号就必须要写，否则 GCC 没办法判断，同样的道理对于其它部分也一样。
这里不会过多展开讲这个技术，详情可参阅GCC 手册。你可以重点看 GAS 相关的章节。
下面将用上面一个函数 atomic_add 为例子说一下，如下所示。
static inline void atomic_add(int i, atomic_t *v)
{
__asm__ __volatile__("lock;" "addl %1,%0"
: "+m" (v->a_count)
: "ir" (i));
}
//"lock;" "addl %1,%0" 是汇编指令部分，%1,%0是占位符，它表示输出、输入列表中变量或表态式，占位符的数字从输出部分开始依次增加，这些变量或者表态式会被GCC处理成寄存器、内存、立即数放在指令中。
//: "+m" (v->a_count) 是输出列表部分，“+m”表示(v->a_count)和内存地址关联
//: "ir" (i) 是输入列表部分，“ir” 表示i是和立即数或者寄存器关联
有了这些原子操作函数之后 ，前面场景中的代码就变成下面这样了：无论有没有中断，或者什么时间来中断，都不会出错。
atomic_t a = {0};
void interrupt_handle()
{
atomic_inc(&a);
}
void thread_func()
{
atomic_inc(&a);
}
好，说完了原子操作，我们再看看怎么用中断控制的思路解决数据并发访问的问题。

方法二：中断控制 搞定复杂变量

中断是 CPU 响应外部事件的重要机制，时钟、键盘、硬盘等 IO 设备都是通过发出中断来请求 CPU 执行相关操作的（即执行相应的中断处理代码），比如下一个时钟到来、用户按下了键盘上的某个按键、硬盘已经准备好了数据。
但是中断处理代码中如果操作了其它代码的数据，这就需要相应的控制机制了，这样才能保证在操作数据过程中不发生中断。
你或许在想，可以用原子操作啊？不过，原子操作只适合于单体变量，如整数。操作系统的数据结构有的可能有几百字节大小，其中可能包含多种不同的基本数据类型。这显然用原子操作无法解决。
下面，我们就要写代码实现关闭开启、中断了，x86 CPU 上关闭、开启中断有专门的指令，即 cli、sti 指令，它们主要是对 CPU 的 eflags 寄存器的 IF 位（第 9 位）进行清除和设置，CPU 正是通过此位来决定是否响应中断信号。这两条指令只能 Ring0 权限才能执行，代码如下。
//关闭中断
void hal_cli()
{
__asm__ __volatile__("cli": : :"memory");
}
//开启中断
void hal_sti()
{
__asm__ __volatile__("sti": : :"memory");
}
//使用场景
void foo()
{
hal_cli();
//操作数据……
hal_sti();
}
void bar()
{
hal_cli();
//操作数据……
hal_sti();
}
你可以自己思考一下，前面这段代码效果如何？
它看似完美地解决了问题，其实有重大缺陷，hal_cli()，hal_sti()，无法嵌套使用，看一个例子你就明白了，代码如下。
void foo()
{
hal_cli();
//操作数据第一步……
hal_sti();
}
void bar()
{
hal_cli();
foo();
//操作数据第二步……
hal_sti();
}
上面代码的关键问题在 bar 函数在关中断下调用了 foo 函数，foo 函数中先关掉中断，处理好数据然后开启中断，回到 bar 函数中，bar 函数还天真地以为中断是关闭的，接着处理数据，以为不会被中断抢占。
那么怎么解决上面的问题呢？我们只要修改一下开启、关闭中断的函数就行了。
我们可以这样操作：在关闭中断函数中先保存 eflags 寄存器，然后执行 cli 指令，在开启中断函数中直接恢复之前保存的 eflags 寄存器就行了，具体代码如下。
typedef u32_t cpuflg_t;
static inline void hal_save_flags_cli(cpuflg_t* flags)
{
__asm__ __volatile__(
"pushfl \t\n" //把eflags寄存器压入当前栈顶
"cli \t\n" //关闭中断
"popl %0 \t\n"//把当前栈顶弹出到flags为地址的内存中
: "=m"(*flags)
:
: "memory"
);
}
static inline void hal_restore_flags_sti(cpuflg_t* flags)
{
__asm__ __volatile__(
"pushl %0 \t\n"//把flags为地址处的值寄存器压入当前栈顶
"popfl \t\n" //把当前栈顶弹出到eflags寄存器中
:
: "m"(*flags)
: "memory"
);
}
从上面的代码中不难发现，硬件工程师早就想到了如何解决在嵌套函数中关闭、开启中断的问题：pushfl 指令把 eflags 寄存器压入当前栈顶，popfl 把当前栈顶的数据弹出到 eflags 寄存器中。
hal_restore_flags_sti() 函数的执行，是否开启中断完全取决于上一次 eflags 寄存器中的值，并且 popfl 指令只会影响 eflags 寄存器中的 IF 位。这样，无论函数嵌套调用多少层都没有问题。

方法三：自旋锁 协调多核心 CPU

前面说的控制中断，看似解决了问题，那是因为以前是单 CPU，同一时刻只有一条代码执行流，除了中断会中止当前代码执行流，转而运行另一条代码执行流（中断处理程序），再无其它代码执行流。这种情况下只要控制了中断，就能安全地操作全局数据。
但是我们都知道，现在情况发生了改变，CPU 变成了多核心，或者主板上安装了多颗 CPU，同一时刻下系统中存在多条代码执行流，控制中断只能控制本地 CPU 的中断，无法控制其它 CPU 核心的中断。
所以，原先通过控制中断来维护全局数据安全的方案失效了，这就需要全新的机制来处理这样的情况，于是就轮到自旋锁登场了。
我们先看看自旋锁的原理，它是这样的：首先读取锁变量，判断其值是否已经加锁，如果未加锁则执行加锁，然后返回，表示加锁成功；如果已经加锁了，就要返回第一步继续执行后续步骤，因而得名自旋锁。为了让你更好理解，下面来画一个图描述这个算法。
自旋锁原理示意图
这个算法看似很好，但是想要正确执行它，就必须保证读取锁变量和判断并加锁的操作是原子执行的。否则，CPU0 在读取了锁变量之后，CPU1 读取锁变量判断未加锁执行加锁，然后 CPU0 也判断未加锁执行加锁，这时就会发现两个 CPU 都加锁成功，因此这个算法出错了。
怎么解决这个问题呢？这就要找硬件要解决方案了，x86 CPU 给我们提供了一个原子交换指令，xchg，它可以让寄存器里的一个值跟内存空间中的一个值做交换。例如，让 eax=memlock，memlock=eax 这个动作是原子的，不受其它 CPU 干扰。
下面我们就去实现自旋锁，代码如下所示。
//自旋锁结构
typedef struct
{
volatile u32_t lock;//volatile可以防止编译器优化，保证其它代码始终从内存加载lock变量的值
} spinlock_t;
//锁初始化函数
static inline void x86_spin_lock_init(spinlock_t * lock)
{
lock->lock = 0;//锁值初始化为0是未加锁状态
}
//加锁函数
static inline void x86_spin_lock(spinlock_t * lock)
{
__asm__ __volatile__ (
"1: \n"
"lock; xchg %0, %1 \n"//把值为1的寄存器和lock内存中的值进行交换
"cmpl $0, %0 \n" //用0和交换回来的值进行比较
"jnz 2f \n" //不等于0则跳转后面2标号处运行
"jmp 3f \n" //若等于0则跳转后面3标号处返回
"2: \n"
"cmpl $0, %1 \n"//用0和lock内存中的值进行比较
"jne 2b \n"//若不等于0则跳转到前面2标号处运行继续比较
"jmp 1b \n"//若等于0则跳转到前面1标号处运行，交换并加锁
"3: \n" :
: "r"(1), "m"(*lock));
}
//解锁函数
static inline void x86_spin_unlock(spinlock_t * lock)
{
__asm__ __volatile__(
"movl $0, %0\n"//解锁把lock内存中的值设为0就行
:
: "m"(*lock));
}
上述代码的中注释已经很清楚了，关键点在于 xchg 指令，xchg %0, %1 。
其中，%0 对应 “r”(1)，表示由编译器自动分配一个通用寄存器，并填入值 1，例如 mov eax，1。而 %1 对应"m"(*lock)，表示 lock 是内存地址。把 1 和内存中的值进行交换，若内存中是 1，则不会影响；因为本身写入就是 1，若内存中是 0，一交换，内存中就变成了 1，即加锁成功。
自旋锁依然有中断嵌套的问题，也就是说，在使用自旋锁的时候我们仍然要注意中断。
在中断处理程序访问某个自旋锁保护的某个资源时，依然有问题，所以我们要写的自旋锁函数必须适应这样的中断环境，也就是说，它需要在处理中断的过程中也能使用，如下所示。
static inline void x86_spin_lock_disable_irq(spinlock_t * lock,cpuflg_t* flags)
{
__asm__ __volatile__(
"pushfq \n\t"
"cli \n\t"
"popq %0 \n\t"
"1: \n\t"
"lock; xchg %1, %2 \n\t"
"cmpl $0,%1 \n\t"
"jnz 2f \n\t"
"jmp 3f \n"
"2: \n\t"
"cmpl $0,%2 \n\t"
"jne 2b \n\t"
"jmp 1b \n\t"
"3: \n"
:"=m"(*flags)
: "r"(1), "m"(*lock));
}
static inline void x86_spin_unlock_enabled_irq(spinlock_t* lock,cpuflg_t* flags)
{
__asm__ __volatile__(
"movl $0, %0\n\t"
"pushq %1 \n\t"
"popfq \n\t"
:
: "m"(*lock), "m"(*flags));
}
以上代码实现了关中断下获取自旋锁，以及恢复中断状态释放自旋锁。在中断环境下也完美地解决了问题。

方法四：信号量 CPU 时间管理大师

无论是原子操作，还是自旋锁，都不适合长时间等待的情况，因为有很多资源（数据）它有一定的时间性，你想去获取它，CPU 并不能立即返回给你，而是要等待一段时间，才能把数据返回给你。这种情况，你用自旋锁来同步访问这种资源，你会发现这是对 CPU 时间的巨大浪费。
下面我们看看另一种同步机制，既能对资源数据进行保护（同一时刻只有一个代码执行流访问），又能在资源无法满足的情况下，让 CPU 可以执行其它任务。
如果你翻过操作系统的理论书，应该对信号量这个词并不陌生。信号量是 1965 年荷兰学者 Edsger Dijkstra 提出的，是一种用于资源互斥或者进程间同步的机制。这里我们就来看看如何实现这一机制。
你不妨想象这样一个情境：微信等待你从键盘上的输入信息，然后把这个信息发送出去。
这个功能我们怎么实现呢？下面我们就来说说实现它的一般方法，当然具体实现中可能不同，但是原理是相通的，具体如下。
1. 一块内存，相当于缓冲区，用于保存键盘的按键码。
2. 需要一套控制机制，比如微信读取这个缓冲区，而该缓冲区为空时怎么处理；该缓冲区中有了按键码，却没有代码执行流来读取，又该怎么处理。
我们期望是这样的，一共有三点。
1. 当微信获取键盘输入信息时，发现键盘缓冲区中是空的，就进入等待状态。
2. 同一时刻，只能有一个代码执行流操作键盘缓冲区。
3. 当用户按下键盘时，我们有能力把按键码写入缓冲区中，并且能看一看微信或者其它程序是否在等待该缓冲区，如果是就重新激活微信和其它的程序，让它们重新竞争读取键盘缓冲区，如果竞争失败依然进入等待状态。
其实以上所述无非是三个问题：等待、互斥、唤醒（即重新激活等待的代码执行流）。
这就需要一种全新的数据结构来解决这些问题。根据上面的问题，这个数据结构至少需要一个变量来表示互斥，比如大于 0 则代码执行流可以继续运行，等于 0 则让代码执行流进入等待状态。还需要一个等待链，用于保存等待的代码执行流。
这个数据结构的实现代码如下所示。
#define SEM_FLG_MUTEX 0
#define SEM_FLG_MULTI 1
#define SEM_MUTEX_ONE_LOCK 1
#define SEM_MULTI_LOCK 0
//等待链数据结构，用于挂载等待代码执行流（线程）的结构，里面有用于挂载代码执行流的链表和计数器变量，这里我们先不深入研究这个数据结构。
typedef struct s_KWLST
{
spinlock_t wl_lock;
uint_t wl_tdnr;
list_h_t wl_list;
}kwlst_t;
//信号量数据结构
typedef struct s_SEM
{
spinlock_t sem_lock;//维护sem_t自身数据的自旋锁
uint_t sem_flg;//信号量相关的标志
sint_t sem_count;//信号量计数值
kwlst_t sem_waitlst;//用于挂载等待代码执行流（线程）结构
}sem_t;
搞懂了信号量的结构，我们再来看看信号量的一般用法，注意信号量在使用之前需要先进行初始化。这里假定信号量数据结构中的 sem_count 初始化为 1，sem_waitlst 等待链初始化为空。
使用信号量的步骤，我已经给你列好了。
第一步，获取信号量。
1. 首先对用于保护信号量自身的自旋锁 sem_lock 进行加锁。
2. 对信号值 sem_count 执行“减 1”操作，并检查其值是否小于 0。
3. 上步中检查 sem_count 如果小于 0，就让进程进入等待状态并且将其挂入 sem_waitlst 中，然后调度其它进程运行。否则表示获取信号量成功。当然最后别忘了对自旋锁 sem_lock 进行解锁。
第二步，代码执行流开始执行相关操作，例如读取键盘缓冲区。
第三步，释放信号量。
1. 首先对用于保护信号量自身的自旋锁 sem_lock 进行加锁。
2. 对信号值 sem_count 执行“加 1”操作，并检查其值是否大于 0。
3. 上步中检查 sem_count 值如果大于 0，就执行唤醒 sem_waitlst 中进程的操作，并且需要调度进程时就执行进程调度操作，不管 sem_count 是否大于 0（通常会大于 0）都标记信号量释放成功。当然最后别忘了对自旋锁 sem_lock 进行解锁。
这里我给你额外分享一个小技巧，写代码之前我们常常需要先想清楚算法步骤，建议你像我这样分条列出，因为串联很容易含糊其辞，不利于后面顺畅编码。
好，下面我们来看看实现上述这些功能的代码，按照理论书籍上说，信号量有两个操作：down，up，代码如下。
//获取信号量
void krlsem_down(sem_t* sem)
{
cpuflg_t cpufg;
start_step:
krlspinlock_cli(&sem->sem_lock,&cpufg);
if(sem->sem_count<1)
{//如果信号量值小于1,则让代码执行流（线程）睡眠
krlwlst_wait(&sem->sem_waitlst);
krlspinunlock_sti(&sem->sem_lock,&cpufg);
krlschedul();//切换代码执行流，下次恢复执行时依然从下一行开始执行，所以要goto开始处重新获取信号量
goto start_step;
}
sem->sem_count--;//信号量值减1,表示成功获取信号量
krlspinunlock_sti(&sem->sem_lock,&cpufg);
return;
}
//释放信号量
void krlsem_up(sem_t* sem)
{
cpuflg_t cpufg;
krlspinlock_cli(&sem->sem_lock,&cpufg);
sem->sem_count++;//释放信号量
if(sem->sem_count<1)
{//如果小于1,则说数据结构出错了，挂起系统
krlspinunlock_sti(&sem->sem_lock,&cpufg);
hal_sysdie("sem up err");
}
//唤醒该信号量上所有等待的代码执行流（线程）
krlwlst_allup(&sem->sem_waitlst);
krlspinunlock_sti(&sem->sem_lock,&cpufg);
krlsched_set_schedflgs();
return;
}
上述代码中的 krlspinlock_cli，krlspinunlock_sti 两个函数，只是对前面自旋锁函数的一个封装，krlschedul、krlwlst_wait、krlwlst_allup、krlsched_set_schedflgs 这几个函数会在进程相关课程进行探讨。

重点回顾

又到了这节课结束的时候，我们回顾一下今天都讲了什么。我把这节课的内容为你梳理一下，要点如下。
1. 原子变量，在只有单个变量全局数据的情况下，这种变量非常实用，如全局计数器、状态标志变量等。我们利用了 CPU 的原子指令实现了一组操作原子变量的函数。
2. 中断的控制。当要操作的数据很多的情况下，用原子变量就不适合了。但是我们发现在单核心的 CPU，同一时刻只有一个代码执行流，除了响应中断导致代码执行流切换，不会有其它条件会干扰全局数据的操作，所以我们只要在操作全局数据时关闭或者开启中断就行了，为此我们开发了控制中断的函数。
3. 自旋锁。由于多核心的 CPU 出现，控制中断已经失效了，因为系统中同时有多个代码执行流，为了解决这个问题，我们开发了自旋锁，自旋锁要么一下子获取锁，要么循环等待最终获取锁。
4. 信号量。如果长时间等待后才能获取数据，在这样的情况下，前面中断控制和自旋锁都不能很好地解决，于是我们开发了信号量。信号量由一套数据结构和函数组成，它能使获取数据的代码执行流进入睡眠，然后在相关条件满足时被唤醒，这样就能让 CPU 能有时间处理其它任务。所以信号量同时解决了三个问题：等待、互斥、唤醒。

思考题

请用代码展示一下自旋锁或者信号量，可能的使用形式是什么样的？
期待你在留言区的分享，也欢迎你把这节课的内容分享给身边的朋友，跟他一起学习交流。
我是 LMOS，我们下节课见！
分享给需要的人，Ta购买本课程，你将得20元
生成海报并分享

赞 69

提建议

© 版权归极客邦科技所有，未经许可不得传播售卖。 页面已增加防盗追踪，如有侵权极客邦将依法追究其法律责任。
上一篇
07 | Cache与内存：程序放在哪儿？
下一篇
09 | 瞧一瞧Linux：Linux的自旋锁和信号量如何实现？
unpreview
 写留言

精选留言(97)

  • pedro
    置顶
    2021-05-26
    今天的专栏可谓是精彩至极！ 锁是解决并发同步问题的关键，从本文来看，锁有两个核心点，一个是原子操作，另一个则是中断； 通过原子操作来实现临界区标志位的改变，关闭中断来避免CPU中途离开导致数据同步失败问题。 自旋锁(spinlock)是锁的最小原型，其它锁都是以它为基础来实现的，自旋锁的实现也颇为简单，只需一个简单的原子标志位就可以实现了，当然还要妥善管理中断。 在 xv6 中，对锁的实现只有两种，一种是刚才提到的 spinlock，而另外一种则是 sleeplock，spinlock 不会让出 CPU 执行权，而 sleeplock 则是在 spinlock 的基础上，增加 sleep 功能，即如果一个执行体(线程或者进程)加锁失败，就会进入休眠状态，让出 CPU 执行权，让其它的任务也能得以执行。 本文中的信号量(sem)也是 sleeplock 的一种，sem 的实现更为精致，通过等待队列来记录加锁失败的执行体，并后续通过一定的策略来选择唤醒，这也是很多编程语言中信号量的实现方式。 当然不同的语言会有不同的优化，比如 go 的 Mutex 是非公平的唤醒机制，但是针对非公平的场景，又设有饥饿补偿，总之本文中实现的 sem 几乎是任何信号量（锁）实现的基础蓝本。 对于思考题答案，这里就顺便贴一下吧，如果有啥问题，欢迎大家交流指正： spinlock_t lock; x86_spin_lock_init(&lock); // 加锁，如果加锁成功则进入下面代码执行 // 否则，一直自旋，不断检查 lock 值为否为 0 x86_spin_lock_disable_irq(&lock); // 处理一些数据同步、协同场景 doing_something(); // 解锁 x86_spin_unlock_enabled_irq(&lock); sem_t sem; x86_sem_init(&sem); // 加锁，减少信号量，如果信号量已经为 0 // 则加锁失败，当前线程会改变为 sleeping 状态 // 并让出 CPU 执行权 krlsem_down(&sem); // 处理一些数据同步、协同场景 doing_something(); // 解锁，增加信号量，唤醒等待队列中的其它线程（若存在） krlsem_up(&sem);
    展开

    作者回复: 我非常喜欢你这样的读者，你总是能抓住问题的本质，继续保持，为你点赞

    共 8 条评论
    130
  • 嗣树
    置顶
    2021-05-26
    记录下我学习本节内容的一些思考： 1. 学习到这里还没有引入抢占这个概念，所以讨论环境中默认进程上下文不会发生抢占，同样也没有进程切换这些东西。 其实讲进程上下文也不对，我们都还没有引入这些东西，暂且凑合这么用吧。 2. 首先讨论了单 cpu 情况下，保证数据一致性的方式，这一时期我们主要防中断： - 对于单个变量我们实现了原子变量（由硬件提供支持），中断是不能打断这个操作的 - 对于复杂变量的操作，这个时候中断可以乱入了，我们通过关中断来保证单 cpu 下这些数据的正确 3. 到了多 cpu 时代，不止有中断这个小三，还有隔壁老王（其他核），关中断已经不管用了， 为了数据的一致需要大家在操作前都走自旋流程，自旋需要新的原子操作支持（xchg），到此我们解决了老王带来的问题。 4. 但是我们还需要面对中断的问题，自旋锁的实现是带条件的死循环，这也引入了一个问题：死锁。 cpu 间的互相抢锁最多抢不到等一会，但是 cpu 和本地中断之间就不同了。 当本 cpu 占有了锁，此时打来中断，假如中断中也要抢这把锁，那他抢不到嘛，只好死给你看咯。 而中断也可以嵌套，这种情况也可能死。所以对自旋锁升级，添加了关中断的操作。 5. 最后老师介绍了信号量，这里其实已经带入了调度的概念。 6. 最后在思考一下抢占和中断优先级带来的问题，其实也还是死锁的问题。 上面我们设想了两种死锁的情况。我们泛化一下，其实中断可以看作比普通进程优先级更高的进程， 只要是构成这种 低级持锁，高级来抢 的局面都可能死锁。而抢占和中断优先级创造了更多的阶级， 也就产生了更多的可能。所以 Linux 中自旋锁的实现第一步是关抢占。 错误或疏漏的地方还请指正，抱拳了老铁。
    展开

    作者回复: 你好，你学的很认真，学的很通透，总结很到位

    共 4 条评论
    34
  • neohope
    置顶
    2021-05-26
    当前版本还有几个问题还没有解决，希望后面课程有进一步详解： 1、跨用户进程时，如何共享内核的同一个锁或信号量 2、没有提供锁的可重入不可重入的限制 3、锁自旋时不会让渡CPU时间 4、暂时没有提供公平锁算法 5、暂时没有提供乐观锁算法 基于本节，其实大家可以尝试一下： 1、信号量如何提供最大资源数限制 2、信号量如何提供扣除多个资源的支持 3、如何实现互斥量这一类数据结构呢 4、如何实现读写锁这一类数据结构呢 锁一般用来做线程间或进程间的互斥操作 信号量一般用来做线程间或进程间资源同步操作，比如资源的占用和释放等
    展开

    作者回复: 是的，我们后面会有介绍 的

    32
  • 惜心（伟祺）
    置顶
    2021-05-27
    这个问题抽象下就是在如何在并行执行中做到串联有序 解决思路就两个： 1.在结果端（确保内存一致） 2.在过程端（确保cpu计算中不被打断） 原子性就是确保内存一致，但是因为cpu计算需要时间所以只能保证一个单位的绝对一致 至于锁、信号都是在对上面两种思路的组合，妙的是会出现跨代组合来实现更多样化的应用 老师从根本上分析问题，由简到繁，一下抓住根本和牛逼 其实操作系统就是对简单的功能的组合重复实现无限复杂操作 这种思路和软件开发是一致的，用一种语言API，各种的组合重复利用时间空间的重复解决各种复杂问题
    展开

    作者回复: 很好，学到了

    12
  • Freddy
    置顶
    2021-05-28
    本节是关于共享数据的并发修改问题，总结了不同场景下的使用方式： 当共享数据是单体变量时，可以尝试使用原子操作指令； 当共享数据是复杂的数据结构时： 当是单CPU环境时，只有中断和业务进程两个代码操作流，此时我们可以手动控制CPU中断关闭/开启，要注意解决CPU中断关闭/开启的嵌套调用问题； 当是多CPU环境时，就不能同时控制多个CPU的中断了，此时我们用到了自旋锁；多个CPU进程竞争自旋锁，成功加锁的进程，可以执行自己的业务流程；这里要注意的是要保证自旋锁流程中的读取锁变量和判断并加锁的操作是原子执行的； 在多CPU环境时，没有获取自旋锁的CPU，就会一直在循环读取锁变量和判断是否加锁的流程当中，浪费了CPU资源，为了解决这个问题，引入了信号量； 首先，各个进程会去竞争信号量； 没有获取信号量的进程放入等待队列，这样该进程所在的CPU就可以去执行其他业务进程了； 获取信号量的进程执行完后，会释放信号量后，同时会去唤醒在等待队列中的进程，这样等待的进程就会再次去竞争信号量； 思考： 1.那一个进程中的多个线程并发修改共享数据的模型，应该也是同样的道理吧？？ 2.锁，应该是操作的前提条件，有了锁才能去执行业务代码； 但对原子操作来说，好像是加锁和业务操作一起执行了。
    展开

    作者回复: 你好，总结到位

    10
  • lifetime
    置顶
    2021-08-18
    看到楼上的各位都发表了那么多的总结，感觉你们都太厉害了！ 我也跟着你们的脚步，谈谈我对这节课的理解： 1、对原子的理解： 抽象理解为一条指令，要么执行完成，要么没有执行 2、对中断的理解： 单个CPU中多个进程并发执行，是靠使用中断进行切换，关闭中断后，只有一个进程在执行这块临界区代码，其他进程无法切换执行，执行完这块代码后，再打开中断，再去切换到别的进程执行 3、对自旋锁的理解： 自旋锁解决多个CPU，多个进程并行执行的情况； 多个进程对同一个物理内存地址进行访问，先访问并判断为0的进程，进程加锁设置为1，执行临界区代码； 其他的进程陷入访问，判断为1的流程中，死循环； 执行完临界区的进程将地址设置为0，其他进程再去争抢 4、对信号量的理解： 信号量解决自旋锁中其他得不到执行的进程一直在轮询的问题，这个一直轮询会导致CPU无法切换到其他不需要执行该临界区的进程执行，效率低下； 所以引入能睡眠的机制，得不到的进程不让他们继续等了，先睡觉，负责其他进程执行的CPU去切换到别的进程执行； 等执行完临界区的进程OK后，再把这些睡觉的进程唤醒，他们再争抢 理解能力有限，只明白这么多，如果有不对的，麻烦指正！
    展开

    作者回复: 对头

    7
  • noisyes
    置顶
    2021-05-27
    真的是把锁的前生今世讲得明明白白，让我有种豁然开朗的感觉！可能到最后可能无法独自实现一个操作系统，但是真的能从底层的角度重新认识操作系统！

    编辑回复: 小编回复，很高兴这个专栏能够帮到你，后面更精彩，敬请期待。还有，只要有兴趣，跟着课程走，是可以跑起来操作系统的，别慌。

    5
  • xiaoq
    置顶
    2022-04-25
    # 单CPU下 业务函数和中断函数会存在并发访问同一资源 1. 对于简单资源(原型变量？)，可以把访问资源变成原子操作，使用带lock前缀的`addl subl incl decl`原子指令； 2. 对于复杂资源(复合类型变量)，可以把访问资源处业务函数关掉中断，保证此处是串行访问资源；需要解决嵌套问题，通过在关中断前使用pushfl、popl保存之前的中断状态，在下一次开启中断时恢复该状态； # 多CPU下 除了每个CPU存在业务函数和中断函数并发问题，还存在不同CPU之间并发问题； 在保证单CPU使用同步的情况下，还需要保证多个CPU同步； 1. 简单资源的原子访问操作：个人理解是因为锁了总线，所以单个、多个CPU均适用； 2. 自旋锁：关键指令xchg，确保 read(if) & set 在多个CPU之间是原子的 # 信号量 在复杂的上下文中保护多个复合资源 使用spinlock和wait_queue以及resource_count实现等待、互斥、唤醒 使用方法 // spinlock spinlock_t spinlock; spinlock_init(&spinlock) // cpu转圈圈，直到获取锁 spinlock_lock(&spinlock) // dosomething() spinlock_unlock(&spinlock) // semaphore sem_t sem; sem_init(&sem); // 线程会休眠 直到拿到信号量 sem_down(&sem); // dosomething() sem_up(&sem);
    展开

    作者回复: 对正确的

    1
  • 不一样的烟火
    2021-07-11
    锁用来保证资源的使用不被打断，打断的情况包括中断，其他cpu执行流，其他线程 保护的情况有原子操作，关中断，自旋锁，信号量 原子操作好比瞬时动作，动作只有一个，不被打断 关中断好比学习的时候关闭电话，不被分心，专心学习 自旋锁好比上厕所的时候锁上门，其他人只能在外面团团转，干着急，其他事儿干不了 信号量好比正在忙，门口挂个闲人免进，等自己办完事儿再通知他过来，他可以先去处理其他事情
    展开

    作者回复: 是的 是的

    共 2 条评论
    12
  • 黎
    2021-05-26
    原子变量，中断，自旋锁，信号量，层层递进，都是在前一个技术的某些场景无法满足的情况下，更高级复杂的解决方案，当然也是基于前面的基础的封装

    作者回复: 你这角度也很透彻，很好

    11
  • Feen
    2021-06-16
    尝试用自己的理解去消化一下本章的内容，用最少的文字在每句话表达出问题和核心和解决办法，不然显得自己都没好好学习： 1、几十年前的一开始，计算机内只有一个CPU，而且执行单道程序，因为当时的整机性能不强，操作方式是一次提交指令等待返回。每次执行任务都发现当前正在运行的程序内部有执行中被打乱的问题，设计程序的时候以为是按照人的思维来运算，实际发现有问题。所以增加了原子操作解决，要不全部执行，要不全部失败。那时候计算机执行任务的时候，程序员都站在计算机（还不能称之为现代电脑）面前，等待结果返回。 2、从各种其他设备开始接入计算机开始，为了解决这些设备也要在计算机运行时候的同时解决输入输出问题（这时CPU开始分时切片运行，表象显得可以同时运行多个程序），引入了中断的概念，CPU整个运行时间被分切为每个单位时间，单位时间之间可以持续运行同一程序，也可以打断运行其他程序，现代操作系统也开始初具雏形，区别与老计算机的标志就在于能否中断。 3、现代电脑蓬勃发展，电脑开始进入千家万户，中断的大量使用给用户提供了无限的可能，单核CPU的频率随着工艺和材料的提升也在提高，在奔腾时代（P2 P3 P4）你甚至可以在主板BIOS中设置中断。 4、万物逃离不了本质，单核CPU频率提升到一定幅度后开始减缓，频率不够，核数来凑。多核CPU进入视野，而中断只能锁定当前占有的那一个CPU核心，无法锁定其他CPU核心不去碰你执行的程序，既然无法通过中断达到锁定，因为你的程序在电脑内只会存在一个程序（如果你一个程序要多开，那通过不同的PID来区分，目前还未讲到，暂且不考虑），聪明的程序员又想到了锁你的程序变量，自旋锁开始使用(xchg)。 5、自旋锁的使用几乎是同时出现信号量，在使用自旋锁的时候确实排他了，但是CPU的工作却发生了变化，不止要执行当前用户程序中正在运行的那一部分代码，还得运行另一部分未执行的代码。同时还有长时间等待的问题。因为它旋转着呢，就如同排队上厕所一样，肯定得敲CPU的大门，问下前面那位占着茅坑的好了没，上不到厕所决不罢休。还好电脑不像人，没有三急来了就骂人的情绪，所以聪明的程序员又干了一件事：发筹子（count），好了叫你，没好的时候躺平。（从这里看出程序的主导权或者执行权，已经由代码本身，移交到外部信号量控制部分。） 6、总结：计算机经过这么多年的发展，从思路上可以看出 ，解决问题的方法就是先从单个示例下手，不管是原子操作，还是中断，还是自旋锁，还是信号量。当你解决了单个实例的问题，那就是从0到1质变，从1到99都是在此基础上的衍生，而从单个实例下手的入口点可以有多个，但从局部性原理开始考虑更轻松一点。
    展开

    作者回复: 嗯 嗯 对的

    共 3 条评论
    8
  • Return12321
    2021-11-01
    信号量理解，比如有微信、QQ、钉钉应用需要获取到键盘输入值： 1、初始化sem_count=1，sem_waitlst等待链为空； 2、通过键盘输入字符到缓冲区，并通过进程调度通知进程； 3、三个进程开始竞争，微信抢到，先给sem_lock加锁，此时QQ、钉钉的进程不能读取信号量，循环判断(自旋锁的逻辑)； 4、微信进程对sem_count执行判断sem_count不小于1(sem_count=1)，获取信号量成功，执行sem_count“减1”操作(sem_count=0)，释放自旋锁sem_lock； 5-1、QQ、钉钉进程此时能读取sem_lock自旋锁，此时钉钉抢到，则先给sem_lock加锁，获取sem_count值并执行判断sem_count不小于1，此时sem_count=0（小于1），钉钉进程进入等待状态，并将其挂入sem_waitlst中，释放自旋锁sem_lock。最后QQ进程重复5-1步骤，sem_count=0，sem_waitlst=[QQ, 钉钉]； 5-2、微信进程执行读取缓存区代码流操作； 6、微信代码流执行(读取键盘值)完成，先给sem_lock加锁，对sem_count执行“加1”操作(sem_count=1)，检查sem_count不小于1，则执行唤醒sem_waitlst 中进程的操作，标记信号量释放成功，给sem_lock解锁； 7、sem_waitlst=[QQ, 钉钉]中的进程重复从4开始的步骤。 文章中写的是先执行“减1”操作、然后sem_count和0判断，但从代码上来看，sem_count是先和1比较，然后执行“减1”操作。
    展开

    作者回复: 是的

    5
  • K菌无惨
    2021-06-13
    感觉信号量好就好在与操作系统的进程/线程调度整合在一起，所以就能够在没有资源时（即信号量等于0），触发进程调度，避免了自旋锁的CPU空转，在拥有资源时能够唤起对应的代码执行下去（即goto start_step）；而至于为什么要用到自旋锁，而不用普通的锁，估计只是因为自旋锁在临界区较短的时候效率更高。 不知道对不对

    作者回复: 对的

    4
  • 李yong
    2021-08-11
    之前没有学习过操作系统，看了两遍，总结一下自己的理解,。 1. 原子操作，可以解决单个变量并发问题。不管后续多核cpu还是单核cpu. 缺点在于只对于单个变量有效。 实现方式，汇编lock. 汇编lock这种方式后续几种方法都会涉及到。 2. 中断。单核cpu时只有一个流，唯一可能的并发就是中断，关闭中断即可解决问题。 缺点，只对于单核cpu有效。 实现方式，关闭中断。 不管单核还是多核，都会存在中断引起并发的情况，所以后面两种方法都涉及关闭中断。 3.自旋锁。可以突破原子操作和中断的限制，即支持多个变量并发以及多核cpu. 额外增加锁变量，锁变量值为0还是1指使此时是否存在并发。存在并发时，循环判断此锁变量的值，直到其他并发结束。不存在并发时，对锁变量加锁。需要关闭中断避免中断的影响。 缺点，存在并发时，其他流一直试图获取锁变量，占用cpu资源。 实现方式，增加锁变量，指使是否存在并发。 4. 信号量。增加结构体，结构体中包含count信号量计数值和list用于挂载等待任务流列表。count为0，任务流进入休眠状态(释放出cpu, 可以调度其他任务流)。count为1.获取信号量，可执行后续操作，并将count置为0，表示已占用，执行完成后，将count加1，
    展开

    作者回复: 对的 你很牛

    3
  • 青玉白露
    2021-07-09
    return (*(volatile u32_t*)&(v)->a_count); 这里为什么非要这么复杂？ return v->a_count; 难道有问题嘛？还得要先取地址再取值

    作者回复: 防止 编译器优化

    3
  • 哇咔咔
    2021-06-06
    “x86 CPU 给我们提供了一个原子交换指令，xchg，它可以让寄存器里的一个值跟内存空间中的一个值做交换” 这里说xchg是原子的，那么为什么在xchg前面还要lock呢？

    作者回复: 当我们用多核cpu时，问题就不同了

    3
  • Geek_0c1f07
    2021-08-22
    中断嵌套问题那里，我想不明白：【开启中断函数】那里到底是怎么才能打开中断的？ 比如嵌套场景如下： 场景：{关闭中断函数，(关闭中断函数，操作数据，开启中断函数)，操作数据，开启中断函数}。 根据代码，只有在关闭中断函数中，能改变内存中的eflags寄存器的标志。而开启中断函数中永远是使用内存中的eflags寄存器的，没有改变内存中的寄存器。那么在此场景中，根据第二个关闭中断函数，内存保存的已经是关闭中断的eflags寄存器，那么开启中断函数每次从内存中读的都是关闭中断的eflags寄存器。所以很矛盾，【开启中断函数】到底是怎么打开中断的？ 望老师解答我的迷惑！
    展开

    作者回复: 因为开启中断有两个方法一个是sti指令，一个是操作eflags寄存器

    2
  • 吴国豪
    2021-07-04
    单核cpu，切换进程必须通过中断，因此可以通过关中断来执行一些同步操作。 多核cpu，同步的数据可能会收到别的核心上的进程影响，因此不能仅仅通过关闭当前核心的中断来执行同步操作。因此通过自旋锁。自旋锁要么被获得，要么就一直循环的申请，直到被获得。 前面的同步操作，都要求锁的使用时间短，否则就会存在多个经常一直在等待某个进程，造成资源的浪费。信号量通过挂起的操作，让获取锁失败的进程，不再进行循环申请的操作，不再占用cpu资源。直到释放锁的进程，在释放的时候，才将挂起的进程释放，让他们进入竞争锁的状态。
    展开

    作者回复: 是的 是的

    2
  • 老男孩
    2021-05-27
    java可以会把锁的状态值放到redis中，然后让redis执行lua脚本来保证执行的原子性。 加锁： if redis.call('setNx',KEYS[1],ARGV[1]) == 1 then if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('expire',KEYS[1],ARGV[2]) else return 0 end else return 0 end 释放锁：if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end 用一个while循环，退出条件是加锁成功。在循环体中把当前线程挂起，并放到一个等待列表中。 获得锁的执行流，最后释放锁，并唤醒等待的线程去竞争锁。 学到这里，我算是明白为什么行业里边有一个鄙视链 ：搞汇编->搞C的->搞C++->搞Java的,看来还是有些道理的。 java有很多类库，工具包，第三方组件来帮助程序员解决偏底层的技术实现，让java程序员更专注业务的开发，也就是java们经常自嘲的堆代码搬砖。 身为一个java老码农，我觉得我也不用纠结了。跟着老师还有大神们认真的学一下，目标不能搞太高，学完这门课至少能从我不知道我不知道变成我知道我不知道就可以了。如果有意外惊喜，那就更好了。这就是所谓的佛系学习法吧。哈哈！
    展开

    作者回复: 是的，完全不用有心理负担，慢慢学，鄙视链完全是心理作用 ，我有时还要写JS、HTML、CSS呢

    2
  • 舞铲阶级
    2021-05-26
    一开始，从单个变量可能由于中断，而产生达不到预期结果的情况，可以通过硬件层面的原子操作，来解决相应问题；但硬件层面的的原子操作没办法适应多种多样的数据结构，这个时候就需要对中断来进行处理，保证数据处理过程中的“原子性”，对于嵌套中断的问题采用栈来保存对应的标志寄存器内容，用于还原来解决嵌套问题；但多核CPU可以绕过单个CPU的中断从而影响数据处理的“原子性”（个人理解希望老师指点），这时候通过硬件提供一个变量来形成自旋锁，保证处理的“原子性”，但是仍会由于嵌套中断从而导致占有自旋锁后进入死循环，因此引入关中断自旋锁来保证“原子性”；最后以上三种情况都不支持长时间等待情况，因此引入了信号量，注意对信号量的操作时要使用自旋锁来保证其“原子性”，如有不对希望大家指正。
    展开

    作者回复: 你好，正确的

    2