并发与保护详解（理论篇）

并发

为了实现同时运行多个程序的需求，人们创造了虚拟化CPU的操作系统，通过在CPU不同的时间片上切换运行不同的程序，产生一种并行的假象，这种假象就是并发

并行 VS 并发：这两者的区别在于，并行是真正意义上的同时运行多个程序，比如多核CPU每个核同时执行不同的程序，而并发是同一个CPU在不同时间段运行不同程序，并不是真正的并行

进程相关

进程，就是一个运行中的程序，是OS为正在运行的程序提供的抽象。

进程有几种基本状态：运行、就绪、阻塞、创建、结束，阻塞挂起、就绪挂起，其中相互转化的关系如下图所示，除就绪态和运行态可相互转化外，其余只能单向转化

就绪状态：可运行，但由于其他进程处于运行状态而暂时停止；针对就绪状态的【就绪挂起】：让暂时没有运行的进程换出到swap space节省内存空间

阻塞状态：等待某一事件而停止运行，就算给了CPU控制权也无法运行；针对阻塞状态的【阻塞挂起】：让阻塞进程在外存等待事件发生解除阻塞状态

挂起状态：阻塞状态的进程会占用物理内存，为了节省内存空间，将该进程的物理内存换出到磁盘中，此时进程不再占有物理内存，称之为挂起状态

进程的管理过程

插叙：进程控制块PCB是一个描述进程的数据结构，包含了进程描述、进程控制和管理信息，以及相关的资源分配清单和CPU相关信息

比如它包括程序计数器，堆栈指针，内存分配状况，所打开文件的状态，用来保证被中断的进程之后能够再次启动，同时它是以链表的形式组织起来的，比如就绪队列、阻塞队列，方便增删管理

• 创建进程：

  • 申请一个空白PCB并填写部分信息；
  • 为该进程分配必需的资源；
  • 将PCB插入到就绪队列，等待被调度运行

• 终止进程，三种方式：正常结束、异常结束、外界干预比如kill掉

  • 查找需要终止的进程的 PCB；
  • 如果处于执行状态，则立即终止该进程的执行，然后将 CPU 资源分配给其他进程；
  • 如果其还有子进程，则应将该进程的子进程交给 1 号进程接管；
  • 将该进程所拥有的全部资源都归还给操作系统；
  • 将其从 PCB 所在队列中删除；

• 阻塞进程，进程需要等待时可以调用阻塞语句把自己阻塞等待，然后等待其他进程唤醒

  • 找到将要被阻塞进程标识号对应的 PCB；
  • 如果该进程为运行状态，则保护其现场，将其状态转为阻塞状态，停止运行；
  • 将该 PCB 插入到阻塞队列中去

• 唤醒进程，唤醒语句和阻塞语句是一一对应的

  • 在该事件的阻塞队列中找到相应进程的 PCB；
  • 将其从阻塞队列中移出，并置其状态为就绪状态；
  • 把该 PCB 插入到就绪队列中，等待调度程序调度；

特殊的进程

• 守护进程：在后台运行的，没有控制终端与它相连的进程。它独立于控制终端，周期性地执行任务，linux的大多数服务器就是用守护进程的方式实现的

  守护进程的创建过程

  1. 让程序在后台执行：调用fork()创建一个子进程，然后让父进程退出
  2. 在子进程中创建新会话：调用setsid()让子进程完全独立
  3. 改变当前目录为根目录（非必选）：调用chdir()防止占用可卸载的文件系统
  4. 重设文件权限掩码（非必选）：调用umask()防止继承的文件创建屏蔽字拒绝某种权限，同时增加守护进程灵活性
  5. 关闭文件描述符（非必选）：节约系统资源
  6. 开始执行守护进程的核心工作

• 僵尸进程：指已经退出、但还未让父进程察觉的子进程，目的是保存子进程的状态，方便父进程在需要的时候获取，弊端就是这个子进程将会持续占据内核资源

  僵尸进程的终止方法

  1. 父进程调用阻塞式的wait()或非阻塞式的waitpid()来终止僵尸子进程，但这会导致父进程挂起
  2. 父进程通过signal函数认为处理信号SIGCHLD：子进程退出后父进程收到信号SIGCHLD，然后在回调函数中调用wait()或waitpid()
  3. 父进程通过signal(SIGCHLD,SIG_IGN)告知内核，自己对子进程的退出不感兴趣，子进程退出后直接由内核回收

线程相关

线程，就是进程当中的一条执行流程，类似于共享地址空间的独立进程

它和进程的一个主要区别是进行上下文切换时不需要切换当前使用的页表（因为同一个进程下的线程们共享地址空间）

插叙：OS事先为CPU准备好寄存器和程序计数器，分别用作缓存以及存储CPU正在执行的指令位置，这是CPU在运行任何任务前所必须依赖的环境，这些环境就叫做CPU上下文

而在OS切换进程时，OS首先会保存当前的上下文信息，然后切换到对应进程的上下文，这个过程就叫做上下文切换，这种切换大致可分为：进程切换、线程切换、中断切换

线程共享资源有：文件描述符表；每种信号的处理方式；当前工作目录；用户ID和组ID

线程非共享资源有：线程id；内核栈；用户空间栈；errno变量；信号屏蔽字；调度优先级

另一个主要区别在于栈，在单线程进程中只有一个栈位于地址空间的底部，而在多线程进程中，每个线程都有一个独立运行的栈

根据线程运行所在区域的不同，可以将线程分为用户线程、内核线程，以及较为特殊的轻量级进程LWP

• 用户线程

  在用户空间里运行的线程，由应用程序创建和管理的线程，所以它也只有用户级别的权限，当一个线程运行结束后必须等待它主动交出CPU控制权或解除

  为什么不能主动打断线程？因为用户空间里并没有中断和切换线程的特权

  插叙：针对这个问题，一方面可以通过非阻塞的IO多路复用技术，比如select、epoll等实现协调多个用户线程之间的切换；另一方面也可以通过获得内核线程的支持，比如后文的1:1模型、N:M模型等实现对用户线程的主动切换

  由于它的切换不涉及到内核态和用户态的切换，切换的速度较快，同时它的实现和管理都和OS没有直接关系，所以可以用于不支持线程技术的OS上

  但由于OS实际上只给【进程】分配了时间片，而它的诞生是进程自作主张地将自己又划分为了多个线程，所以每个线程获得的时间片都较小，多线程运行下的速度较慢，也无法充分利用多核CPU的优势

  操作系统只看得见【进程】，看不见进程划分出的用户线程，所以无法多核同时执行该进程下的多条线程，除非获得内核线程的支持

• 内核线程

  由OS管理它的创建、终止和调度的线程

  可由OS直接将时间片分配给它，于是多线程的进程可以获得更多的CPU运行时间，且由于内核有强制中断、切换和调度线程的特权，所以如果某个线程被阻塞了是不需要等待它主动交出CPU的

  但也由于它的创建切换和终止都是通过系统调用实现的，所以系统开销会比较大

• 轻量级进程LWP

  可以将它理解为是内核支持的用户线程，它与内核线程一对一映射，可以通过它为用户线程提供内核的支持

  根据用户线程和它的数量对应关系，可以分为如下三种模型：

  • 1:1模型：一个LWP对应一个用户线程

    可实现并行，当一个LWP阻塞时，不会影响到其他的LWP；但每一个用户线程都会跟一个内核线程对应，创建线程的开销会比较大

  • N:1模型：一个LWP对应多个用户线程

    线程管理在用户态下完成，该模式中的用户线程对操作系统是不可见的；上下文切换时的效率较快，但无法充分利用多核也无法主动切换一个阻塞的用户线程

  • M:N模型：多个LWP对应多个用户线程

    综合了以上两种的优点，大部分情况下的线程上下文切换发生在用户空间，效率较高；也能较为充分地利用多核CPU的优势

单个进程如何运行

在我们了解了进程和线程的相关概念后，让我们开始讨论一个重要的问题，那就是为了构建让多个任务共享CPU同时运行的假象，我们需要在保持操作系统对CPU的控制权的情况下，尽可能地获得高性能

这就需要用到接下去讨论的受限直接运行机制

这个机制顾名思义分为两个部分：一个部分是”直接执行“，另一个部分是”受限“

直接运行：OS启动程序时创建好进程，然后跳转到进程入口点开始执行用户代码，直接在CPU上运行程序，用来实现高性能的需求

受限：但上面的这种直接运行需要加以一定的限制，防止OS失去对进程们的控制权，但在某些情况下又必须要允许它进行一些特权操作

比如在一些情况下，需要允许进程进行磁盘I/O或申请内存等操作

为此，操作系统引入了一种新的处理器模式：用户模式和内核模式

内核模式下运行的代码则可以执行所有类型的受限操作

而用户模式下用户运行的代码将受到限制，但操作系统允许内核小心地向用户程序暴露一些关键功能（也就是系统调用），让用户可以通过系统调用执行特权操作

执行特权操作的过程：

1. 用户使用系统调用执行trap指令跳入内核中并提高特权级别
2. 用户在内核中执行用户所期望的工作代码
3. 当用户完成工作后OS执行return-from-trap指令
4. 回到用户程序并降低特权级别

执行特权操作需要的操作系统支持：

1. OS在启动时需要通过特权指令初始化陷阱表（trap table），通过陷阱表告诉硬件在发生异常事件时需要运行哪些代码
2. 由CPU记住陷阱表的位置
3. 硬件必须确保有足够的寄存器存储调用者的信息，以便OS执行return-from-trap指令后能够返回到正确的位置

为什么需要陷阱表呢？：如果由调用者告诉trap指令应该执行内核里的哪段代码，会使得程序员可以令内核运行任意代码序列，这将破坏操作系统的安全性

整个受限直接运行的流程图如下

多个进程如何并发

前文提到，并发并不是真正的并行，而是通过CPU在不同时间片上切换执行不同的进程（线程）给各个进程（线程）营造出一种并行的假象，这里就涉及到了两个主要的问题：如何切换进/线程 以及 在什么时候切换进/线程

如何切换进/线程？

先以进程为例，当OS决定切换进程时，它将为正在执行的进程保存一些资源

比如说虚拟内存、栈、全局变量等用户空间资源以及内核堆栈、寄存器等内核空间资源，此处为OS明确地保存，这些资源指的就是【上下文】

并为即将执行的进程恢复一些寄存器的值，通过切换栈使得当内核进入【切换代码调用】时是当前进程的上下文，而最后执行return-from-trap时将返回跳入到新进程的上下文中，从而实现进程的切换

进程切换过程：保存当前进程上下文到进程PCB中，并从下一个进程PCB取出上下文恢复到CPU中

这就要求进程在运行结束后将CPU的控制权交由给OS，才能让OS得以进行上下文切换，一般来说有两种方式让进程转交控制权

1. 协作方式：当OS相信进程会合理运行时，它通过等待系统调用或某种非法操作的发生后，从而重获CPU控制权

2. 非协作方式——时钟中断：当OS没有进程协作时，它在启动时通过特权指令启动时钟，并将时钟编程为每隔一段时间就产生中断，当发生中断时硬件将保存运行进程的寄存器到内核栈，而OS重新获得CPU的控制权，它可以选择切换进程，也可以选择继续运行该进程

   OS将在启动时就设置好硬件在发生时钟中断时需要调用的代码，从而让硬件得以顺利保存寄存器

   此处为硬件的隐式保存，区别于前文OS明确地保存寄存器

而对于线程来说，不同进程下的线程切换与上文所述的进程上下文切换没有差别；同一个进程下的线程切换也十分类似，只是由于这些线程共享虚拟内存和全局变量等资源，在切换时只切换私有数据、寄存器等数据，开销较小

线程是调度的基本单位，进程是资源拥有的基本单位，所以OS在进行任务调度时，实际上是在调度线程，而进程只是给线程提供资源，所以严格上来说，进程切换就是不同进程下的线程切换

在什么时候切换进/线程？

这其实就是在问如何调度多个需要运行的进/线程，或者说进/线程间的调度算法是什么，当进程的状态发生变化的时候，都会触发OS的调度，比如说当进程从就绪态变成运行态时，OS会从就绪进程队列中挑选出一个来运行

如果硬件提供周期性的时钟中断，根据如何处理中断可以将调度算法分成两类

抢占式算法：进程只运行一段时间，到点了就挂起通过中断把CPU控制权返回给调度程序调度；

非抢占式算法：让进程运行到直到自己阻塞或退出才会调用其他程序

为了衡量不同进程调度侧策略的优劣，有如下几个指标

• 周转时间：T周转=T完成 -T到达，如果有n个进程需要完成，T平均周转=(T周转1+T周转2+……)/n，应该避免进程等待时间很长，而运行时间很短，以降低周转时间

• 公平性：应该避免一些任务长时间占据CPU，而另一些任务却始终无法执行被“饿死”

  然而周转时间和公平性两个性能追求往往是矛盾的，为了进程间的公平，就要求OS在一些时间段阻止当前进程运行切换到另一些进程，这就提高了当前进程的周转时间

• 响应时间：T响应=T任务到达-T首次运行，对于交互性比较强的应用，应该尽可能缩短它的响应时间

• 系统吞吐量：权衡长任务和短任务的运行完成数量

• CPU利用率：进程因发生IO事件阻塞的时候应该及时切换进程，充分利用CPU

FIFO先进先出策略

顾名思义，先就绪的进程先开始执行（First-In-First-Out），适用于CPU繁忙的系统

优点：简单且易于实现；当所有的任务时间差距不大时，表现较好

缺点：当最先执行的进程时间大大超过之后的进程，会导致之后的进程等待太多时间，T平均周转被拉高

SJF最短任务优先策略

在FIFO策略的基础上，如果多个进程同时到达，优先执行耗时短的进程，有利于提高系统吞吐量，但对长任务不友好

缺点：如果多个进程到达的时间不同，若最先到达的进程耗时最长，依旧会拉高T平均周转

STCF最短完成时间策略

每当新工作进入系统时，它会确定剩余工作和新工作中，谁的剩余完成时间最少，然后抢占调度执行可以最快完成的工作

比如说一开始OS正在执行A进程，预计需要工作10min，这个时候来了进程B和C，这两个进程预计只需要工作1min

那么OS将通过中断获得CPU控制权，切换执行进程B（因为他的所需完成时间最短），执行完B之后将继续切换执行C，最后才返回进程A执行

优点：这个策略就可以较好地解决FIFO策略和SJF策略在周转时间上的缺陷，保证每次都让短任务优先被处理掉

缺点：但如果只使用这种策略，将会导致响应时间被延长，对交互性特别不友好

以上面那种情况为例，A的响应时间为0，B的响应时间也为0（B刚到达就开始运行），而C进程的响应时间为1min（想象一下，你按下键盘一分钟之后电脑屏幕才出现你按的字符）

MLFQ多级反馈队列策略

在前文的讨论中，我们可以发现，降低周转时间的核心在于优先运行短任务

但不论是SJF策略还是STCF策略，想到达到优先运行短任务这个目的，都需要预先知道各进程可能的工作时间，然而这却是OS难以得知的，且以上策略在交互性方面的表现都不甚理想

而MLFQ多级反馈队列策略则很好地解决了这两个问题

首先在MLFQ中有许多优先级不同的独立队列，每个队列下有多个工作

MLFQ通过观察这些工作执行时的行为动态地调整它们的优先级，并且总是先轮转执行最高优先级队列下的工作，其余具体规则如下：

• 当新工作进入系统时，放在最高优先级队列

• 当工作用完整个时间片后，降低其优先级移入到下一个队列

  目的：降低周转时间，短任务将在高优先级队列被很快处理，而长任务将被降低优先级，下放到低优先级队列

• 若工作在其时间片内主动释放CPU，则保持优先级不变

  目的：提高交互性，让不断出让CPU的进程保持在高优先级队列，实现交互性工作的快速运行

• 一旦工作用完了它在某一级队列的时间配额，无论放弃了多少次CPU都降级

  目的：保护OS，防止有进程总是在时间片结束前执行一次I/O，霸占CPU

• 每经过一段时间S，就将系统中所有工作重新加入最高优先级队列

  目的：保证公平性，防止有太多交互性工作或短任务不断占用高优先级队列，导致低优先级队列里的长任务被“饿死”

保护

当我们通过进程、线程抽象了正在运行的程序，也通过诸如受限直接运行机制、多级反馈机制等初步建立了计算机的并发问题之后，一切都还没有结束，如何在并发的情况下保证程序不出错变成了一个新的难题

考虑这么一种情况：我们创建了A和B两个线程，在两个线程中都让一个全局变量nums自增1000次，我们预期的nums最终结果将会是2000，然而当我们实际这么做后会发现计算机给出的答案往往并不符合我们的预期，这就是因为缺乏了对并发的保护

为什么会出错？：如果在线程A运行过程中发生了不合时宜的中断，比如A线程让nums变量变成1之后，还没有来得及将nums值保存到内存地址中，却立刻发生了中断切换到了B线程；

B线程从内存地址中取出nums的值，由于A线程还没有更新保存，此时取到的nums仍为0，B线程将它自增变成1后存回地址中；

随后A线程恢复运行，它并不知道在B线程所发生的事，而只是继续执行自己的最后一步：将自增为1的nums保存到内存地址中，然而在预期情况下，这个时候变量应该是2才对

像这样导致多个程序访问共享资源的代码叫做临界区，多个执行线程都试图更新共享资源的情况叫做竞态条件，当程序由一个或多个竞态条件组成的时候，程序的输出常常因运行而已，导致出现结果的不确定性，而我们期望计算机给出确定的结果

之所以会出现这种不确定性，关键在于有多个进/线程同时访问了临界区，且产生了不合时宜的中断，所以我们在一个时间段内应该只允许一个线程进入临界区操作共享资源，这就是互斥的概念

想要实现这种互斥，一般来说有两种方式，一种是：锁，另一种是：信号量

锁

锁，是一种变量，它有着多种状态，比如：available、free状态代表它可用，而held、acquire状态代表它正在被一个线程持有。

当一段临界区代码被上锁了之后，就只有获取了锁的线程可以进入临界区，而其他试图访问临界区的线程都会因为获取锁失败而阻塞，从而实现了线程间的互斥

最常见的使用方式如下：

lock_t mutex;//锁是一种变量，使用前需要先声明
...
lock(&mutex);//给临界区加上锁，当某个线程运行到此处时会尝试获取锁
arr++;       //临界区代码，只有持有锁的线程可以访问此处的代码
unlock(&mutex);//解锁，当某个线程运行到此处会释放掉它所持有的锁
lock_t mutex;//锁是一种变量，使用前需要先声明
...
lock(&mutex);//给临界区加上锁，当某个线程运行到此处时会尝试获取锁
arr++;       //临界区代码，只有持有锁的线程可以访问此处的代码
unlock(&mutex);//解锁，当某个线程运行到此处会释放掉它所持有的锁
lock_t mutex;//锁是一种变量，使用前需要先声明
...
lock(&mutex);//给临界区加上锁，当某个线程运行到此处时会尝试获取锁
arr++;       //临界区代码，只有持有锁的线程可以访问此处的代码
unlock(&mutex);//解锁，当某个线程运行到此处会释放掉它所持有的锁

OS使用了一些硬件原语来实现锁，比如说TestAndSet(测试并设置指令)，CompareAndSwap(比较并交换指令)，以及FetchAndAdd(获取并增加指令)，下面我们逐个剖析这三种硬件原语是如何实现锁的

硬件原语：一组特定的计算机硬件指令，在硬件层面将一些操作设置为原子性的，比如TestAndSet，将【测试】与【设置】两个操作设置为原子操作

TestAndSet如何实现锁

int TestAndSet(int *old_ptr,int new){
    int old=*old_ptr;
    *old_ptr=new; //将old_ptr设置为新值 
    return old;   //返回old_ptr的旧值
}
​
typedef struct lock_t{
    int flag;
}lock_t;
​
void init(lock_t lock){
    lock->flag=0;
}//初始化锁
​
void lock(lock_t lock){
    while(TestAndSet(&lock->flag,1)==1) ;//空等待，自旋
}
​
void unlock(lock_t *lock){
    lock->flag=0;
}
int TestAndSet(int *old_ptr,int new){
    int old=*old_ptr;
    *old_ptr=new; //将old_ptr设置为新值 
    return old;   //返回old_ptr的旧值
}


​



typedef struct lock_t{
    int flag;
}lock_t;
​
void init(lock_t lock){
    lock->flag=0;
}//初始化锁
​

void lock(lock_t lock){
    while(TestAndSet(&lock->flag,1)==1) ;//空等待，自旋
}

​
void unlock(lock_t *lock){
    lock->flag=0;
}
int TestAndSet(int *old_ptr,int new){
    int old=*old_ptr;
    *old_ptr=new; //将old_ptr设置为新值 
    return old;   //返回old_ptr的旧值
}


​



typedef struct lock_t{
    int flag;
}lock_t;
​
void init(lock_t lock){
    lock->flag=0;
}//初始化锁
​

void lock(lock_t lock){
    while(TestAndSet(&lock->flag,1)==1) ;//空等待，自旋
}

​
void unlock(lock_t *lock){
    lock->flag=0;
}

• 当flag为0值时说明锁可被获取，flag值为1时说明锁正在被人持有

• 当线程获取锁时：调用TestAndSet函数，由于flag值为0，所以它返回锁的旧值，会跳出while循环，同时设置flag的值为1；当线程离开临界区时，调用unlock()将flag清理为0

• 而当一个线程尝试获取不可得的锁时，由于锁的值一直是1，TestAndSet函数返>回值一直都是1，所以它会一直在while循环里自旋

  在lock()函数中的while循环下，如果什么都不做，就说明线程因获取锁失败而阻塞的时候，会一直等待，这种锁被叫做自旋锁；而如果在while循环下把线程信息保存进TCB，并切换线程，这就变成了无等待锁（？互斥锁吧）

CompareAndSwap如何实现锁

int CompareAndSwap(int *ptr,int expected,int new){
    int actual=*ptr;
    if(actual==expected) *ptr=new;
    return actual;
}
​
void lock(lock_t *lock){
    while(CompareAndSwap(&lock->flag,0,1)==1) ;
}
int CompareAndSwap(int *ptr,int expected,int new){
    int actual=*ptr;
    if(actual==expected) *ptr=new;
    return actual;
}


​



void lock(lock_t *lock){
    while(CompareAndSwap(&lock->flag,0,1)==1) ;
}
int CompareAndSwap(int *ptr,int expected,int new){
    int actual=*ptr;
    if(actual==expected) *ptr=new;
    return actual;
}


​



void lock(lock_t *lock){
    while(CompareAndSwap(&lock->flag,0,1)==1) ;
}

• 检测ptr是否与expected相等（即检测flag值是否为0，即锁可否被获取）
• 如果相等，则更新ptr的值为新值，然后返回ptr的旧值（如果返回的是0，说明线程在尝试获取锁的时候flag是0，可获取该锁；如果返回的是1，说明已经有线程占有该锁了，就会在while循环里自旋等待

FetchAndAdd如何实现锁

int FetchAndAdd(int *ptr){
    int old=*ptr);
    *ptr=old+1;
    return old;
}
​
typedef struct lock_t{
    int ticket;
    int turn;
}lock_t;
​
void lock_init(lock_t *lock){
    lock->ticket=0;
    lock->turn=0;
}
​
void lock(lock_t *lock){
    int myturn= FetchAndAdd(&lock->ticket);
    while(lock->turn!=myturn) ;
}
​
void unlock(lock_t *lock){
    FetchAndAdd(&lock->turn);
}
int FetchAndAdd(int *ptr){
    int old=*ptr);
    *ptr=old+1;
    return old;
}


​



typedef struct lock_t{
    int ticket;
    int turn;
}lock_t;
​
void lock_init(lock_t *lock){
    lock->ticket=0;
    lock->turn=0;
}
​
void lock(lock_t *lock){
    int myturn= FetchAndAdd(&lock->ticket);
    while(lock->turn!=myturn) ;
}
​
void unlock(lock_t *lock){
    FetchAndAdd(&lock->turn);
}
int FetchAndAdd(int *ptr){
    int old=*ptr);
    *ptr=old+1;
    return old;
}


​



typedef struct lock_t{
    int ticket;
    int turn;
}lock_t;
​
void lock_init(lock_t *lock){
    lock->ticket=0;
    lock->turn=0;
}
​
void lock(lock_t *lock){
    int myturn= FetchAndAdd(&lock->ticket);
    while(lock->turn!=myturn) ;
}
​
void unlock(lock_t *lock){
    FetchAndAdd(&lock->turn);
}

• FetchAndAdd的作用：将ptr自增，然后返回它的旧值
• 每次调用lock函数会使得ticket值自增，每次调用unlock函数会使得turn值自增
• 在lock函数里，如果ticket自增前的旧值与unlock相等，说明现在没有线程在持有锁（ticket的值和turn相等，说明调用lock函数的次数和unlock函数的次数是一样）

使用这种硬件原语实现的锁可以保证让所有线程都能抢到锁，只要一个线程获得了ticket值，那它最终就必然会被调度

不同类型的锁

通过以上方式，在lock函数里使用while() ;让线程空等待的锁都是自旋锁，加锁失败的线程将会自旋等待，直到锁可用为止

需要注意的是，如果在单核CPU上想要使用自旋锁，必须搭配抢占式的调度器，因为一个自旋的线程永远不会主动放弃CPU控制权，这将会导致这个单核CPU一直在自旋

这种锁在多核系统中一般不会主动产生线程切换，适合异步、协程等在用户态切换请求的编程方式

然而如果有太多的线程竞争一把锁，将会导致自旋过多，浪费CPU资源，因此我们需要对自旋锁进行一定的改进：当一个线程加锁失败后，将其休眠，释放CPU资源并切换到其他线程，直到该线程所需要的锁被释放后才重新唤醒它，这就是互斥锁

然而这种锁也存在一定的开销的成本，它会涉及到两次的线程上下文切换

第一次：当前线程加锁失败，切换到其他线程；第二次：锁可用后唤醒原线程，从其他线程切换回它

所以使用这种锁需要注意，它适用于被锁住的代码执行时间较长，如果执行时间较短的话，它单单是上下文切换的事件可能都比执行时间要更长

综合以上考虑，实际上，在Linux系统中采用了一种古老的锁方案：两阶段锁，OS设计者意识到自旋在处理很快就要释放锁的场景下效率较高，因此规定两阶段锁的第一阶段将会自旋一段时间，如果自旋阶段加锁失败，才会让调用线程休眠

针对一些特定的应用场景，比如明确区分读写共享资源操作的场景下，还有读写锁的概念，实现的核心功能为：当写锁被线程持有时，将会阻塞其他线程获取读锁和写锁的操作；而当写锁未被持有时，可有多个线程同时持有读锁

即允许多个用户同时读取共享资源；且当有一个用户正在写资源时，不允许其他用户同时写或同时读，以防发生错误

根据实现不同，还可以分为读优先锁和写优先锁

• 读优先锁：当有线程调用写锁被阻塞时，该线程需要等待所有的读锁都被释放，才能加写锁成功
• 写优先锁：当有线程调用写锁被阻塞时，不管期间是否有多少线程又调用读锁，都会优先让该线程加写锁成功，阻塞其他线程加读锁的操作

但这两种方式其实都有一些问题，都有可能让读锁或写锁饿死，于是就又产生了公平读写锁，用队列把获取锁的线程排队，不论是读锁还是写锁都按先入先出的原则加锁即可

锁的其他分类：

悲观锁：互斥锁、自旋锁、读写锁都属于是悲观锁：即认为多线程同时修改共享资源的概率很高，所以访问共享资源前要先上锁

乐观锁：先修改共享资源，再验证一下这段时间是否有其他线程在修改资源，若有则放弃本次操作，若没有则操作完成：即认为多线程同时修改资源的概率是比较小的。常应用于共享文档、Git等场景：总是让用户先编辑内容，在提交的时候通过版本号来判断是否发生冲突，这避免了加锁解锁的操作，但冲突重试的成本较高

信号量

信号量是一个有整数值的对象，可由初始值决定其行为，有三个主要的操作函数，分别是用于初始化信号量sem的sem_init()，执行P操作的sem_post()，以及执行V操作的sem_wait()

• P操作：将信号量sem自增1，如果此时有线程正在等待唤醒，则唤醒一个等待线程
• V操作：将信号量sem自减1，如果自减后信号量<0，则阻塞当前线程进入休眠，等待其他线程执行P操作唤醒它

应用场景1：实现互斥锁

一般来说，信号量的初始值sem设置为多少，就意味着允许多少个线程在同一时间内使用这个信号量，根据这个原则，可以将sem初始值设置为1实现互斥，这样的信号量叫做二值信号量（信号量只有持有 和 没有持有两种状态）

最常见的使用方式如下：

sem_t m;
sem_init(&m,0,1);
……
sem_wait(&m);
arr++;
sem_post(&m);
sem_t m;
sem_init(&m,0,1);
……
sem_wait(&m);
arr++;
sem_post(&m);
sem_t m;
sem_init(&m,0,1);
……
sem_wait(&m);
arr++;
sem_post(&m);

为什么信号量的初始值需要设置为1？

假设有两个线程想要进入临界区，线程1先调用sem_wait()将信号量减一变成0（相当于获取锁），它会继续往下走进入临界区；

如果现在线程2也想进入临界区调用sem_wait()，信号量再减一变成负数，会让线程2开始等待；

直到线程1结束了临界区代码调用sem_post()将信号量加一，才会唤醒线程2（相当于释放锁）

应用场景2：实现读写锁

typedef struct _rwlock_t{
    sem_t lock;
    sem_t writelock;
    int readers;
}rwlock_t;
​
void rwlock_init(rwlock_t *rw){
    rw->readers=0;
    sem_init(&rw->lock,0,1);
    sem_init(&rw->writelokc,0,1);
}
​
void rwlock_acquire_readlock(rwlock_t *rw){
    sem_wait(&rw->lock);
    rw->readers++;
    if(rw->readers==1) sem_wait(&rw->writelock);
    //只要正在执行读操作的用户达到了1，就通过sem_wait()将写者锁阻塞
    sem_post(&rw->lock);
}
​
void rwlock_release_readlock(rwlock_t *rw){
    sem_wait(&rw->lock);
    rw->readers--;
    if(rw->readers==0) sem_post(&rw->writelock);
    //只要没有用户正在执行读操作，就通过sem_post()将写者锁释放
    sem_post(&rw->lock);
}
​
void rwlock_acquire_writelock(rwlock_t *rw){
    sem_wait(&rw->writelock);
}
​
void rwlock_release_writelock(rwlock_t *rw){
    sem_post(&rw->writelock);
}
typedef struct _rwlock_t{
    sem_t lock;
    sem_t writelock;
    int readers;
}rwlock_t;
​



void rwlock_init(rwlock_t *rw){
    rw->readers=0;
    sem_init(&rw->lock,0,1);
    sem_init(&rw->writelokc,0,1);
}
​
void rwlock_acquire_readlock(rwlock_t *rw){
    sem_wait(&rw->lock);
    rw->readers++;
    if(rw->readers==1) sem_wait(&rw->writelock);
    //只要正在执行读操作的用户达到了1，就通过sem_wait()将写者锁阻塞
    sem_post(&rw->lock);
}
​
void rwlock_release_readlock(rwlock_t *rw){
    sem_wait(&rw->lock);
    rw->readers--;
    if(rw->readers==0) sem_post(&rw->writelock);
    //只要没有用户正在执行读操作，就通过sem_post()将写者锁释放
    sem_post(&rw->lock);
}
​
void rwlock_acquire_writelock(rwlock_t *rw){
    sem_wait(&rw->writelock);
}
​
void rwlock_release_writelock(rwlock_t *rw){
    sem_post(&rw->writelock);
}
typedef struct _rwlock_t{
    sem_t lock;
    sem_t writelock;
    int readers;
}rwlock_t;
​



void rwlock_init(rwlock_t *rw){
    rw->readers=0;
    sem_init(&rw->lock,0,1);
    sem_init(&rw->writelokc,0,1);
}
​
void rwlock_acquire_readlock(rwlock_t *rw){
    sem_wait(&rw->lock);
    rw->readers++;
    if(rw->readers==1) sem_wait(&rw->writelock);
    //只要正在执行读操作的用户达到了1，就通过sem_wait()将写者锁阻塞
    sem_post(&rw->lock);
}
​
void rwlock_release_readlock(rwlock_t *rw){
    sem_wait(&rw->lock);
    rw->readers--;
    if(rw->readers==0) sem_post(&rw->writelock);
    //只要没有用户正在执行读操作，就通过sem_post()将写者锁释放
    sem_post(&rw->lock);
}
​
void rwlock_acquire_writelock(rwlock_t *rw){
    sem_wait(&rw->writelock);
}
​
void rwlock_release_writelock(rwlock_t *rw){
    sem_post(&rw->writelock);
}

• lock变量：用于实现一个互斥锁，在rwlock_acquire_readlock()和rwlock_release_lock()两个函数中，保证增减readers数量 和 阻塞、释放写者锁两个操作是原子式的

• writelock变量：用于限制写者锁只有在没有用户操作共享资源时才能获取

• readers变量：用于记录正在进行读操作的用户有几个

  为什么是if(rw->readers==1) sem_wait(rw->writelock);，而不是if(rw->readers>=1) sem_wait(rw->writelock);：因为想要阻塞写者锁的获取，只需要让rw->writelock为-1即可，如果是后者的写法，每次获取写者锁都会让rw->writelock自减，是不必要的操作

  以如上方式实现的读写锁是读优先锁，需要等待rw->readers==0，也就是没有用户在读时才会让进程获取写锁，如果想要实现读写公平锁则需要增加一个额外的flag信号量，用来限制读者或写者无限涌入的优先特权

应用场景3：用作条件变量，实现多线程间的同步

首先介绍一下什么是同步：并发进程或线程在一些关键点上可能需要互相通信和等待，比如需要等待线程A结束后才能进行线程B，这种互相制约的等待和互通信息就是同步

比如说主线程需要等待子线程中的一些条件成立，才能继续运行，这个用作条件的变量就是条件变量，想要实现这种条件变量，需要将信号量设置为0

条件变量本身不是锁，它用来自动阻塞一个线程，等待直到某个特殊情况发生为止

sem_t s;
void child(void *arg){
    ……
    sem_post(&s);//将s值自增1,作为一种信号传递给主线程，告知子线程已执行结束
    return NULL;
}
​
int main(int argc,char* argv[]){
    sem_init(&s,0,0);
    ……
    pthread_t c;
    Pthread_create(c,NULL,child,NULL); //创建一个线程
    sem_wait(&s);
    ……
    return 0;
}
sem_t s;
void child(void *arg){
    ……
    sem_post(&s);//将s值自增1,作为一种信号传递给主线程，告知子线程已执行结束
    return NULL;
}


​


int main(int argc,char* argv[]){
    sem_init(&s,0,0);
    ……
    pthread_t c;
    Pthread_create(c,NULL,child,NULL); //创建一个线程
    sem_wait(&s);
    ……
    return 0;
}
sem_t s;
void child(void *arg){
    ……
    sem_post(&s);//将s值自增1,作为一种信号传递给主线程，告知子线程已执行结束
    return NULL;
}


​


int main(int argc,char* argv[]){
    sem_init(&s,0,0);
    ……
    pthread_t c;
    Pthread_create(c,NULL,child,NULL); //创建一个线程
    sem_wait(&s);
    ……
    return 0;
}

为什么要将信号量设置为0？：保证主线程如果先于子线程调用时，执行到sem_wait()就会被阻塞，只有等待子线程执行结束后调用sem_post()将信号量增为1才能继续执行

接下来，让我们考虑一种更复杂的情况：生产者-消费者问题，也就是生产者将资源生产放到缓冲区，然后消费者从缓冲区取走资源的场景，比如HTTP中的数据缓冲区就是一个典型的生产者-消费者问题

想要解决这个问题，总共需要三个信号量：

0. 互斥信号量mutex，初始值设为1，用于互斥访问缓冲区，也就是在生产和取走数据的前后加上sem_post(&mutex)和sem_wait(&mutex)
1. 资源信号量full，初始值为0，用于表示缓冲区有多少数据可供消费者取走，生产者生产资源后调用sem_wait(&full)让full值自增，而消费者消费资源前调用sem_post(&full)先判断一下有没有资源可以取，没有则阻塞等待
2. 资源信号量empty，初始值为MAX，用于表示缓冲区有多少空位可供生产者放东西，生产者生产资源前先调用sem_post(&empty)检查是否有空位，消费者消费后调用sem_wait(&empty)让empty值自增

具体代码如下：

sem_t empty;
sem_t full;
sem_t mutex;
​
void *producer(void* arg){
    for(int i=0;i<loops;i++){
        sem_wait(&empty);//V操作先判断一下是否有空位
        sem_wait(&mutex);
        put(i);
        sem_post(&mutex);
        sem_post(&full);//P操作告知被阻塞的消费者进程，有资源可以取了
    }
}
​
void *consumer(void *arg){
    for(int i=0;i<loops;i++){
        sem_wait(&full);
        sem_wait(&mutex);
        int tmp=get();
        sem_post(&mutex);
        sem_post(&empty);
        printf("%d\n",tmp);
    }
}
sem_t empty;
sem_t full;
sem_t mutex;
​
void *producer(void* arg){
    for(int i=0;i<loops;i++){
        sem_wait(&empty);//V操作先判断一下是否有空位
        sem_wait(&mutex);
        put(i);
        sem_post(&mutex);
        sem_post(&full);//P操作告知被阻塞的消费者进程，有资源可以取了
    }
}

​

void *consumer(void *arg){
    for(int i=0;i<loops;i++){
        sem_wait(&full);
        sem_wait(&mutex);
        int tmp=get();
        sem_post(&mutex);
        sem_post(&empty);
        printf("%d\n",tmp);
    }
}
sem_t empty;
sem_t full;
sem_t mutex;
​
void *producer(void* arg){
    for(int i=0;i<loops;i++){
        sem_wait(&empty);//V操作先判断一下是否有空位
        sem_wait(&mutex);
        put(i);
        sem_post(&mutex);
        sem_post(&full);//P操作告知被阻塞的消费者进程，有资源可以取了
    }
}

​

void *consumer(void *arg){
    for(int i=0;i<loops;i++){
        sem_wait(&full);
        sem_wait(&mutex);
        int tmp=get();
        sem_post(&mutex);
        sem_post(&empty);
        printf("%d\n",tmp);
    }
}

为什么可以把sem_wait(&mutex)放在sem_wait(&full)或sem_wait(&empty)之前，sem_post(&mutex)放在sem_post(&empty)或sem_post(&full)之后，扩大互斥域吗？

不行！假设消费者获取了mutex，取走了数据后调用sem_post()唤醒了生产者，但生产者想要写入数据前需要等待mutex，而此刻持有mutex的消费者却在等待full，变成了死锁

应用场景4：解决“哲学家就餐”问题

什么是哲学家就餐问题？：几个哲学家围着一个圆桌，每2个哲学家之间有一把餐叉，每个哲学家时而思考时而吃饭，而只有左右手同时拿到餐叉的哲学家才能吃饭

这个问题的挑战在于，如何保证没有死锁，没有哲学家饿死，且尽可能让更多哲学家同时吃饭（也就是实现更高的并发度）

最简单的方案就是：假设餐叉数组为forks[N]，给每个餐叉都设置一个初始值为1的信号量，在每个哲学家线程中都先后使用sem_post(forks[i]); sem_post(forks[(i+1)%N]);意为先让哲学家先尝试拿起左手边的餐叉，再尝试拿起右手边的餐叉，然后即可就餐

缺陷：然而这种方案下如果产生了不合时宜的中断，比如说每个哲学家都只执行了sem_post(forks[i])就被中断去执行其他哲学家线程，这会导致每个哲学家都只拿起了一个餐叉，每个人都没得吃

这个方案的缺陷在于，每个哲学家线程里都编程设定每个人都先同一手边的餐叉，比如都统一先拿左手边的餐叉。所以针对这个方案的改进也很简单：只要让每个哲学家线程获取餐叉的先后顺序交错一下，比如说规定奇数位次的哲学家都先拿左手边在拿右手边，偶数次位的哲学家相反，这样就可以避免原方案的极端情况出现。

常见的并发问题

对于并发中常出现的问题，一般来说可以分为两类，第一类占据了并发问题的多数：非死锁问题，第二类就是死锁问题

非死锁问题

0. 违反原子性：代码段本意是原子的，但执行中没有强制实施原子性，如下

   Thread 1::
   if (thd->proc_info){
       ...
       fputs(thd->proc_info,...);
       ...
   }
   ​
   Thread 2::
   thd->proc_info=NULL;
   Thread 1::
   
   if (thd->proc_info){
       ...
   
   
       fputs(thd->proc_info,...);
       ...
   }
   
   
   ​
   
   
   Thread 2::
   thd->proc_info=NULL;
   Thread 1::
   
   if (thd->proc_info){
       ...
   
   
       fputs(thd->proc_info,...);
       ...
   }
   
   
   ​
   
   
   Thread 2::
   thd->proc_info=NULL;
   

   • 如果线程1在fputs之前被线程2中断，thd->proc_info被置为NULL，再进行线程1程序将崩溃
   • 修改该类问题：只需要在访问共享变量前加锁，访问后解锁即可

1. 违反顺序缺陷：代码中A应该在B之前执行，但并没有强制实现这种顺序性

   Thread 1::
   void init(){
       ...
       mThread = CreateThread(mMain,...);
       ...
   }
   ​
   Thread 2::
   void mMain(){
       ...
       mState = mThread->State;
       ...
   }
   Thread 1::
   void init(){
       ...
   
   
       mThread = CreateThread(mMain,...);
       ...
   }
   
   
   ​
   
   
   Thread 2::
   void mMain(){
       ...
       mState = mThread->State;
       ...
   }
   Thread 1::
   void init(){
       ...
   
   
       mThread = CreateThread(mMain,...);
       ...
   }
   
   
   ​
   
   
   Thread 2::
   void mMain(){
       ...
       mState = mThread->State;
       ...
   }
   
   

   • 如果线程2在线程1之前执行，则mThread并没有被初始化就被使用，可能会出现奇怪的问题

   • 修改该类问题：使用条件变量，比如如下修改

     Thread 1::
     void init(){
         ...
         mThread = CreateThread(mMain,...);
         sem_post(&m);
         ...
     }
     ​
     Thread 2::
     void mMain(){
         ...
         sem_t m;
         sem_init(&m,0,1);
         sem_wait(&m);
         mState = mThread->State;
         ...
     }
     Thread 1::
     
     void init(){
         ...
     
     
         mThread = CreateThread(mMain,...);
         sem_post(&m);
         ...
     }
     ​
     Thread 2::
     void mMain(){
         ...
         sem_t m;
         sem_init(&m,0,1);
         sem_wait(&m);
         mState = mThread->State;
         ...
     }
     Thread 1::
     
     void init(){
         ...
     
     
         mThread = CreateThread(mMain,...);
         sem_post(&m);
         ...
     }
     ​
     Thread 2::
     void mMain(){
         ...
         sem_t m;
         sem_init(&m,0,1);
         sem_wait(&m);
         mState = mThread->State;
         ...
     }
     
     

死锁问题

什么是死锁问题？

举个栗子，我们有线程1和线程2两个工作线程，线程1持有了锁A后准备获取锁B，但是它被线程2中断了，线程2快一步持有了锁B，而它接下来却要等待锁A；从而导致两个线程在互相等待的情况，这就是一种典型的死锁问题

通过上面的例子，我们可以发现，死锁问题的诞生有四个必要条件：

0. 互斥：即线程所需要的资源是互斥的，比如锁
1. 持有并等待：即线程持有了某个资源，又在等待其他资源，比如线程1抢到了锁A，等待锁B
2. 非抢占：即线程已获得的资源除非它主动放弃，否则不会被其他线程抢占
3. 循环等待：即线程之间存在等待资源的环路，比如每个线程都持有一个锁，而这个锁由恰好是它接下去的一个线程所等待的

这四个条件只要一个不满足，就不会产生死锁问题，所以针对这四个条件，就有不同的死锁预防策略：

0. 防止互斥：通过强大的硬件指令，构造出不需要锁的数据结构，比如说通过先前的TestAndSet就可以实现两个操作的原子式执行

1. 防止持有并等待：原子式地抢锁，保证任何线程在抢占任何锁前需要先抢到全局的prevention锁，比如线程1在持有了锁A后，就算被线程2中断，线程2也会因为没有prevention锁而无法获取锁B，从而让线程1可以顺利获取锁A和B

2. 防止非抢占：也就是让线程获取资源多次失败后，主动放弃已经获取的资源后再重试，比如线程1第二次尝试获取锁B失败后，将会释放锁A，从而线程2可以顺利获得锁A进行其后的工作

   但这种情况可能会导致活锁问题：多个线程一直在重复执行这个释放-重试的工作，但彼此都同时抢锁失败，解决方法是：让线程结束一次释放-重试操作后先随机等待一段时间再重复该操作

3. 防止循环等待：这也是最直接的一种针对策略：在获取锁时提供一个全序，让线程每次申请锁时都严格按照固定的顺序，比如让线程1和线程2都是先获取锁A，再获取锁B，这样就算线程1被中断切换到线程2，线程2也无法获取锁A，不会产生死锁

除了死锁预防以外，在某些场景也许更适合死锁避免，也就是当我们了解到需要在n个处理器上进行m个进程任务时，可以通过进程调度避免死锁，比如避免让例子中的线程1和线程2并发运行，虽然这一定程度上会牺牲了性能

而如果已经发生了死锁，还有一些死锁恢复方案，主要有如下三种：

• 利用抢占恢复：挂起进程强行取走资源给另一个进程
• 利用回滚恢复：复位到更早还没有取得所需资源的状态
• 利用kill恢复：杀死循环等待环路中的一个或多个进程