@TOC

零、前言

本章主要讲解学习Linux中对多线程的执行中的互斥与安全问题

一、Linux线程互斥

1、基本概念及引入

• 互斥相关概念：

1. 临界资源：多线程执行流共享的资源就叫做临界资源

2. 临界区：每个线程内部，访问临界资源的代码，就叫做临界区

3. 互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用

4. 原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成

• 示例：模拟抢票

#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
int thickets=100;//100张票
//thickets--表示抢票
void* Routine(void* arg)
{
    while(1)
    {
        if(thickets>0)
        {
            usleep(30000);//抢票时间
            printf("%s get a thickets, now thickets' number:%d\n",(char*)arg,--thickets);
        }
        else 
            break;
    }
    return NULL;
}

int main()
{
    pthread_t tid1,tid2,tid3;
    pthread_create(&tid1,NULL,Routine,(void*)"thread 1");
    pthread_create(&tid2,NULL,Routine,(void*)"thread 2");
    pthread_create(&tid3,NULL,Routine,(void*)"thread 3");

    pthread_join(tid1,NULL);
    pthread_join(tid2,NULL);
    pthread_join(tid3,NULL);
    return 0;
}

• 效果：

注：变量tickets被多个执行流同时访问，所以thickets就是一个临界资源，当访问临界资源时，判断tickets是否大于0、打印剩余票数以及--tickets的代码也就是临界区

• 出现负数的原因：

if语句判断条件为真以后，代码可以并发的切换到其他线程

usleep用于模拟漫长业务的过程，在这个漫长的业务过程中，可能有很多个线程会进入该代码段

–ticket操作本身就不是一个原子操作，可能在执行当中也被切换成其他线程

• 具体可能的过程：

当thickets为1时，一个线程进行if判断为真，进入代码段，当执行到usleep进行系统调用休眠，返回时到用户态时线程发生切换，多个线程此时也进行if判断为真（thickets还是1），这些线程当进行打印的时候进行了多次的减减操作，也就造成了负数的情况

• – 操作并不是原子操作，而是对应三条汇编指令：

1. load ：将共享变量ticket从内存加载到寄存器中

2. update : 更新寄存器里面的值，执行-1操作

3. store ：将新值，从寄存器写回共享变量ticket的内存地址

• –执行对应的汇编代码：

152 40064b: 8b 05 e3 04 20 00 mov 0x2004e3(%rip),%eax # 600b34 <ticket>
153 400651: 83 e8 01 sub $0x1,%eax
154 400654: 89 05 da 04 20 00 mov %eax,0x2004da(%rip) # 600b34 <ticket>

注：因为减减操作并不是原子的，当减减操作第一步执行完（thickets=100），可能该线程的时间片到了（寄存器中的数据被保存eax=100），其他线程切入，而切入的线程执行了多次减减并写会到内存（thickets=80），当切出的线程切回时，恢复线程上下文数据（eax=100），再进行减减（eax=99），把数据写回到内存时（thickets=99），此时的数据的值只达到了一次减减的效果，此时的资源并不安全

2、互斥量mutex介绍

• 概念：

1. 大部分情况，线程使用的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况变量归属单个线程，其他线程无法获得这种变量

2. 但有时候，很多变量都需要在线程间共享，这样的变量成为共享变量，可以通过数据的共享，完成线程之间的交互

3. 多个线程并发的操作共享变量，就会带来一些问题

• 要解决以上问题需要做到三点：

1. 代码必须要有互斥行为：当代码进入临界区执行时，不允许其他线程进入该临界区

2. 如果多个线程同时要求执行临界区的代码，并且临界区没有线程在执行，那么只能允许一个线程进入该临界区

3. 如果线程不在临界区中执行，那么该线程不能阻止其他线程进入临界区

注：要做到这三点，本质上就是需要一把锁，Linux上提供的这把锁叫互斥量

• 示图：

3、互斥量的使用

• 初始化互斥量：

1. 静态分配

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER

2. 动态分配

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrictattr);

参数：mutex：要初始化的互斥量；attr：互斥量的属性，一般设置为NULL

• 销毁互斥量：

int pthread_mutex_destroy(pthread_mutex_t *mutex)；

• 注意：

1. 使用 PTHREAD_ MUTEX_ INITIALIZER 初始化的互斥量不需要销毁

2. 不要销毁一个已经加锁的互斥量

3. 已经销毁的互斥量，要确保后面不会有线程再尝试加锁

• 互斥量加锁和解锁:

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

返回值：成功返回0，失败返回错误号

• 调用 pthread_ lock 时可能遇到的情况:

1. 互斥量处于未锁状态，该函数会将互斥量锁定，同时返回成功

2. 发起函数调用时，其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量，那么pthread_ lock调用会陷入阻塞(执行流被挂起)，等待互斥量解锁

• 示例：改进抢票

#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
int thickets=100;//100张票
//thickets--表示抢票
pthread_mutex_t lock;//线程共用一个互斥锁
void* Routine(void* arg)
{
    while(1)
    {
        pthread_mutex_lock(&lock);
        if(thickets>0)
        {
            usleep(100000);//抢票时间
            printf("%s get a thickets, now thickets' number:%d\n",(char*)arg,--thickets);
            pthread_mutex_unlock(&lock);
        }
        else
        {
            pthread_mutex_unlock(&lock);
            break;
        }
        usleep(100000);
    }
    return NULL;
}
int main()
{
    pthread_mutex_init(&lock,NULL);
    pthread_t tid1,tid2,tid3;
    pthread_create(&tid1,NULL,Routine,(void*)"thread 1");
    pthread_create(&tid2,NULL,Routine,(void*)"thread 2");
    pthread_create(&tid3,NULL,Routine,(void*)"thread 3");

    pthread_join(tid1,NULL);
    pthread_join(tid2,NULL);
    pthread_join(tid3,NULL);
    pthread_mutex_destroy(&lock);
    return 0;
}

• 效果：

4、互斥量原理

• 概念：

1. 对于互斥锁来说被多个线程同时可见，也就是说互斥锁本身就是一个临界资源，所以互斥锁想要保护临界区的互斥性，那么互斥锁操作则一定是原子的

2. 为了实现互斥锁操作，大多数体系结构都提供了swap或exchange指令，该指令的作用是把寄存器和内存单元的数据相交换，由于只有一条指令，保证了原子性

3. 即使是多处理器平台，访问内存的总线周期也有先后，一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期

• 示图：伪代码

注：在交换和赋值的过程中本质就是让竞争的多线程中保证中有一个线程的交换得到的寄存器数据为1，即保证同一时刻只有一个竞争的线程为1，由此才能往下执行，否则只能进行等待

二、可重入/线程安全

1、基本概念

• 线程安全：

1. 多个线程并发同一段代码时，不会出现不同的结果，没有数据错乱的情况

2. 常见对全局变量或者静态变量进行操作，并且没有锁保护的情况下，会出现该问题

• 重入：

1. 同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入

2. 一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则是不可重入函数

• 注意：

1. 对于可重入来说是函数的特性，对于线程安全来说是线程的特性

2. 如果一个函数是可重入的，那么执行还函数的多线程是线程安全的

2、线程安全

• 常见线程不安全的情况：

1. 不保护共享变量的函数

2. 函数状态随着被调用，状态发生变化的函数

3. 返回指向静态变量指针的函数

4. 调用线程不安全函数的函数

• 常见的线程安全的情况：

1. 每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的

2. 类或者接口对于线程来说都是原子操作

3. 多个线程之间的切换不会导致该接口的执行结果存在二义性

3、重入函数

• 常见不可重入的情况：

1. 调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的

2. 调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构

3. 可重入函数体内使用了静态的数据结构

• 常见可重入的情况：

1. 不使用全局变量或静态变量

2. 不使用用malloc或者new开辟出的空间

3. 不调用不可重入函数

4. 不返回静态或全局数据，所有数据都有函数的调用者提供

5. 使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据

4、联系与区别

• 可重入与线程安全联系：

1. 函数是可重入的，那就是线程安全的

2. 函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题

3. 如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的

• 可重入与线程安全区别：

1. 可重入函数是线程安全函数的一种

2. 线程安全不一定是可重入的，而可重入函数则一定是线程安全的

3. 如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数若锁还未释放则会产生死锁，因此是不可重入的

三、常见锁概念

• 死锁：

死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态

• 死锁四个必要条件：

1. 互斥条件：一个资源每次只能被一个执行流使用

2. 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放

3. 不剥夺条件:一个执行流已获得的资源，在末使用完之前，不能强行剥夺

4. 循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系

注：对于死锁，四个条件缺一不可

• 避免死锁：

1. 破坏死锁的四个必要条件

2. 加锁顺序一致

3. 避免锁未释放的场景

4. 资源一次性分配

• 避免死锁算法：

1. 死锁检测算法

2. 银行家算法

四、Linux线程同步

1、基本概念

• 同步概念与竞态条件：

1. 同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步

2. 竞态条件：因为时序问题，而导致程序异常，我们称之为竞态条件

• 注意：

1. 在多线程中，为了保护临界资源，我们需要用到互斥锁，但是在线程竞争的情况下，此外我们还需要考虑资源的一些特殊情况

2. 在特殊的情况下，可能存在某个线程多次的竞争获取锁，但是却没有做出实际的事情，这种频繁的申请虽然没有什么问题，但是不是很合理

3. 同时如果线程的竞争力非常强，这就可能导致其他线程长时间竞争不到锁，引起饥饿问题

4. 由此我们需要对于这种特殊的情况，保证线程能够按照某种次序进行临界资源的访问，由此就需要条件变量

• 条件变量：

当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。例如一个线程访问队列时，发现队列为空，它只能等待，只到其它线程将一个节点添加到队列中

2、条件变量的使用

• 初始化条件变量：

1. 静态分配

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

2. 动态分配

• 初始化函数原型：

int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrictattr);

• 解释：

1. 参数：cond：要初始化的条件变量；attr：设置属性，一般填NULL

2. 返回值：条件变量初始化成功返回0，失败返回错误码

• 销毁函数原型：

int pthread_cond_destroy(pthread_cond_t *cond)

• 解释：

1. 参数：cond：需要销毁的条件变量

2. 返回值：条件变量销毁成功返回0，失败返回错误码

3. 使用PTHREAD_COND_INITIALIZER初始化的条件变量不需要销毁

• 等待条件满足函数原型：

int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);

• 解释：

1. 功能：进行等待直到条件符合被唤醒

2. 参数：cond：需要等待的条件变量；mutex：当前线程所处临界区对应的互斥锁

3. 返回值：函数调用成功返回0，失败返回错误码

• 唤醒等待函数原型：

int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

• 解释：

1. 区别：pthread_cond_signal函数用于唤醒等待队列中首个线程；pthread_cond_broadcast函数用于唤醒等待队列中的全部线程

2. 参数：cond：唤醒在cond条件变量下等待的线程

3. 返回值：函数调用成功返回0，失败返回错误码

• 示例：协同调度其他线程

#include<stdio.h>
#include<pthread.h>
#include<unistd.h>

pthread_cond_t cond;
pthread_mutex_t mutex;
void* Routine1(void* arg)
{
    //被调度线程执行
    while(1)
    {
        pthread_cond_wait(&cond,&mutex);
        printf("%s is running...\n",(char*)arg);
    }
}
void* Routine2(void* arg)
{
    int cnt=0;
    while(1)
    {
        if(cnt%3!=0)
            pthread_cond_signal(&cond);
        else 
            pthread_cond_broadcast(&cond);
        cnt++;
        sleep(1);
    }
}

int main()
{
    pthread_cond_init(&cond,NULL);
    pthread_mutex_init(&mutex,NULL);

    pthread_t tid1,tid2,tid3,tid4;
    pthread_create(&tid1,NULL,Routine1,(void*)"thread 1");
    pthread_create(&tid2,NULL,Routine1,(void*)"thread 2");
    pthread_create(&tid3,NULL,Routine1,(void*)"thread 3");
    pthread_create(&tid4,NULL,Routine2,(void*)"thread 4");

    pthread_join(tid1,NULL);
    pthread_join(tid2,NULL);
    pthread_join(tid3,NULL);
    pthread_join(tid4,NULL);
    
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
    return 0;
}

• 效果：

3、条件变量等待

• 为什么条件变量等待函数第二个参数需要互斥锁：

1. 条件等待是线程间同步的一种手段，如果只有一个线程，条件不满足，一直等下去都不会满足，所以必须要有一个线程通过某些操作，改变共享变量，使原先不满足的条件变得满足，并且友好的通知等待在条件变量上的线程
2. 条件不会无缘无故的突然变得满足了，必然会牵扯到共享数据的变化，所以一定要用互斥锁来保护，没有互斥锁就无法安全的获取和修改共享数据
3. 进入访问临界资源时，申请互斥锁，当遇到条件变量等待时，传入第二个参数互斥锁，等待的同时会将所申请到的互斥锁给释放，被唤醒的时候会同时将互斥锁给竞争上，保证数据安全

• 示图：

注：如果不释放互斥锁，那么其他线程无法成功申请到锁进而改变数据，也就没有办法通知等待的线程，那么申请到锁的线程一直等待，别的线程无法获取锁也无法通知，也就会造成死锁

• 错误伪代码设计：访问临界资源时，先上锁，发现条件不满足，解锁，然后等待在条件变量上

pthread_mutex_lock(&mutex);
while (condition_is_false) 
{
    pthread_mutex_unlock(&mutex);
    //解锁之后，等待之前，条件可能已经满足，信号已经发出，但是该信号可能被错过
    pthread_cond_wait(&cond);
    pthread_mutex_lock(&mutex);
} 
pthread_mutex_unlock(&mutex);

• 注意：

1. 这里由于解锁和等待不是原子操作。调用解锁之后， pthread_cond_wait 之前，如果已经有其他线程获取到互斥量，并且条件满足，发送了唤醒信号，那么 pthread_cond_wait 将错过这个信号，可能会导致线程永远阻塞在这个 pthread_cond_wait ，所以解锁和等待必须是一个原子操作

2. 调用pthread_cond_wait函数会去看条件量是否等于0：如果等于，就把互斥量改为1，直到cond_ wait返回，把条件量改成1，把互斥量恢复成原样，也就是不满足条件时，在进行等待前，把互斥锁给解锁，当等待到被唤醒时会自动竞争到互斥锁

4、条件变量使用规范

1. 等待条件代码

pthread_mutex_lock(&mutex);
while (条件为假）{
    pthread_cond_wait(cond, mutex);
}
//修改条件
pthread_mutex_unlock(&mutex);

注：这里可能存在被伪唤醒的情况，当唤醒的时候可能竞争锁失败，继续等待，其他线程竞争成功执行后并释放锁，此时条件判断为假，但是该线程竞争到锁后会继续往下执行，如果没有再次进行判断可能造成错误，使用while循环判断保证醒来后条件一定为真才往下走

2. 给条件发送信号代码

pthread_mutex_lock(&mutex);
//设置条件为真
pthread_cond_signal(cond);
pthread_mutex_unlock(&mutex);

五、POSIX信号量

1、信号量概念及介绍

• 基本概念：

1. POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步

2. 信号量本质是一个描述临界资源中资源数目的计数器，信号量能够更细粒度的对临界资源进行管理，每个执行流在进入临界区之前都应该先申请信号量，申请成功就有了访问临界资源的权限，当访问离开就进行释放信号量（类似一个访问预定机制）

3. 一般来说我们是将临界资源作为一个整体看待，所以需要使用互斥锁让同一时刻只能有一个执行流进行访问临界资源；实际对于临界资源我们可以选择分割为多个区域，当多个执行流需要访问不同区域的临界资源时，那么我们可以让这些执行流同时访问临界资源的不同区域，此时不会出现数据不一致等问题

• 信号量的PV操作：

1. P操作：申请信号量获得临界资源中某块资源的使用权限，当申请成功时逻辑上临界资源中可使用资源数目减一，对应到信号量上就是让计数器减一

2. V操作：释放信号量归还临界资源中某块资源的使用权限，当释放成功时逻辑上临界资源中可使用的资源数目加一，对应到信号量上就是让计数器加一

• 注意：

1. 信号量本质也是临界资源（被多个执行流申请），要保护临界资源所以信号量的PV操作必须是原子操作

2. 当临界资源申请完时，信号量为0，再申请时线程会在该信号量的等待队列当中进行等待，直到有信号量被释放时再被唤醒

• 二元信号量：

1. 如果将信号量的初始值设置为1，那么此时该信号量叫做二元信号量

2. 信号量的初始值为1，说明信号量所描述的临界资源只有一份，此时信号量的作用基本等价于互斥锁

2、信号量的使用

• 初始化信号量函数原型：

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);

• 解释：

1. 参数：sem：需要初始化的信号量；pshared:0表示线程间共享，非零表示进程间共享；value：信号量初始值

2. 返回值：初始化信号量成功返回0，失败返回-1

• 销毁信号量函数原型：

int sem_destroy(sem_t *sem);

• 解释：

参数：sem：需要销毁的信号量

返回值：销毁信号量成功返回0，失败返回-1

• 等待信号量函数原型：

int sem_wait(sem_t *sem); //P()

• 解释：

1. 功能：等待信号量，会将信号量的值减1

2. 参数：sem：需要等待的信号量

3. 返回值：等待信号量成功返回0，信号量的值减一；等待信号量失败返回-1，信号量的值保持不变

• 发布信号量函数原型：

int sem_post(sem_t *sem);//V()

• 解释：

1. 功能：发布信号量，表示资源使用完毕可以归还资源了，将信号量值加1

2. 参数：sem：需要发布的信号量

3. 返回值：发布信号量成功返回0，信号量的值加一；发布信号量失败返回-1，信号量的值保持不变

• 示例：

#include <iostream>
#include <string>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
class Sem{
public:
	Sem(int num)
	{
		sem_init(&_sem, 0, num);
	}
	~Sem()
	{
		sem_destroy(&_sem);
	}
	void P()
	{
		sem_wait(&_sem);
	}
	void V()
	{
		sem_post(&_sem);
	}
private:
	sem_t _sem;
};
Sem sem(1); //二元信号量
int tickets = 2000;
void* TicketGrabbing(void* arg)
{
	std::string name = (char*)arg;
	while (true){
		sem.P();
		if (tickets > 0){
			std::cout << name << " get a ticket, tickets left: " << --tickets << std::endl;
			sem.V();
            usleep(1000);
		}
		else{
			sem.V();
			break;
		}
	}
	std::cout << name << " quit..." << std::endl;
	pthread_exit((void*)0);
}
int main()
{
	pthread_t tid1, tid2, tid3, tid4;
	pthread_create(&tid1, nullptr, TicketGrabbing, (void*)"thread 1");
	pthread_create(&tid2, nullptr, TicketGrabbing, (void*)"thread 2");
	pthread_create(&tid3, nullptr, TicketGrabbing, (void*)"thread 3");
	pthread_create(&tid4, nullptr, TicketGrabbing, (void*)"thread 4");
	
	pthread_join(tid1, nullptr);
	pthread_join(tid2, nullptr);
	pthread_join(tid3, nullptr);
	pthread_join(tid4, nullptr);
	return 0;
}

• 效果：