Linux_线程

• 线程与进程
• 多级页表
• 线程控制
• 线程互斥
• 线程同步
• 生产者消费者模型
• 常见概念

下面选取32位系统举例。

一.线程与进程

上图是曾经我们认为进程所占用的资源的集合。

1.1 线程概念

1. 线程是一个执行分支，执行粒度比进程细，调度成本比进程低
2. 线程是cpu调度的基本单位，进程是分配资源的基本单位
3. 线程是进程内部的一个执行流

通常我们创建线程是为了执行程序的一部分代码，所以执行粒度一定比进程更细，我们知道进程=内核数据结构+代码和数据。引入线程之后，这个概念就应该修正为进程=许多内核数据结构+代码和数据。这些数据结构指向同一个程序地址空间，如图：

在操作系统的概念中，这样的数据结构叫做线程控制块（TCB），但由于创建新的数据结构还要设计新的调度逻辑，所以Linux在内核中并没有设计这样的数据结构，而是复用进程的数据结构PCB。故此在Linux中并没有真正的线程，而是用进程模拟的线程，线程在Linux中叫轻量级进程（LWP）。

1.2 理解线程

1. 创建线程只需要执行一部分代码，所以执行粒度比进程细
2. 当发生调度时，操作系统会识别是进程间的调度还是线程间的调度，如果是线程间的调度，那么只需要改变部分寄存器的值即可，而cache不需要重新加载代码。所以线程的调度成本是比进程小的。
3. cpu每次调度都是以线程为单位的，操作系统分配资源是以进程为单位的。
4. 线程有自己的栈
5. 线程共享进程的大部分数据，但也有私有数据
   1. **线程ID **
   2. **一组寄存器 **
   3. **栈 **
   4. **errno **
   5. **信号屏蔽字 **
   6. 调度优先级
      

1.3 线程优点

• 创建一个新线程的代价要比创建一个新进程小得多
• 与进程之间的切换相比，线程之间的切换需要操作系统做的工作要少很多
• 线程占用的资源要比进程少很多
• 能充分利用多处理器的可并行数量
• 在等待慢速I/O操作结束的同时，程序可执行其他的计算任务
• 计算密集型应用，为了能在多处理器系统上运行，将计算分解到多个线程中实现
• I/O密集型应用，为了提高性能，将I/O操作重叠。线程可以同时等待不同的I/O操作。
  

1.4 线程缺点

• 健壮性减低：主线程退出，全部线程退出，一个线程收到信号，所有线程收到信号
• 缺乏访问控制：相比与进程，线程之间的通信成本是极低的，但是线程之间的通信缺乏访问控制，即同步机制。当一个线程正在写入，还没有写完，另一个线程就可能会读走数据。
  

二.多级页表

文件系统指出磁盘和内存的数据交换是以块（4KB）为单位，因此内存的管理也要以4KB为单位，在内存中这样的结构叫做页。

如果页表只有一张，并且页表保存的是字节间的映射关系，那么一个页表需要保存2^{32行，这样肯定是不行的，一是查找速度慢，二是内存可能不够。因此，Linux中采用多级页表，并且将32虚拟地址拆分成3部分使用：前10位用来在页目录中使用，中间10位在页表项中使用，最后12位做为页内偏移地址使用。（2}12 = 4KB）

页目录共有1024行，每个页表项也有1024行，共可以映射2^20页地址，这恰好是内存中页的数量，页表映射出物理页号，物理页号和虚拟地址中的12位页内偏移组合，构成物理字节地址。

由于程序不可能使用整个内存，所以页表不会一次全部创建，而是创建一部分。

三.线程控制

3.1 线程库的理解

在Linux中，没有真正意思上线程，所以系统不会提供线程相关的接口，但是有控制轻量级进程的相关接口。可是用户只认线程，所以有了用户级线程库pthread，这个库底层封装了轻量级进程的相关接口，给上层提供了线程相关的接口。pthread库在任何一个Linux系统上都自带了，因此pthread库也叫做原生线程库。由于pthread库属于第三方库，所以在使用gcc/g++编译时，需要指定库名称。gcc xxxx -lpthread

3.2 创建线程

• pthread_creat
• pthread_t：整数，输出型参数，返回给用户线程id，以后操作线程用线程id。
• attr：可以定义线程的属性，一般设置为空
• void*(start_routine)(void) ：函数指针，设置线程指向的函数
• void* arg：线程执行函数的参数
• 返回值：成功返回0，失败返回-1

代码示例：

#include<iostream>
#include<pthread.h>
#include <unistd.h>using namespace std;void* threadRoutine(void* s)
{const char* str = (const char*)s;cout << str << endl;}
int main()
{pthread_t p1;pthread_create(&p1, nullptr, threadRoutine, (void*)"thread 1");sleep(3);return 0;
}

通过命令ps -aL可以查看当前用户创建的线程，线程不分父子，分主次：主线程的PID=LWP。操作系统判断是进程间切换还是线程间切换通常使用PID来区分的，看两个结构体的PID是否相同，相同即是线程间切换。

• 如果主线程退出，当前进程的所有线程立即退出。
• 如果任一线程收到信号，那么所有线程都会收到信号。

3.3 线程等待

如果主线程不等待新线程，那么新线程会产生类似僵尸进程的情况，导致内存泄漏。

• pthread_join

• 等待指定id的线程
• void** retval：输出型参数，已知线程执行的任务返回值为void*，因此要获得这个值就需要void*的地址，即void**

代码示例：

#include<iostream>
#include<pthread.h>
#include <unistd.h>
#include <cstring>using namespace std;void* threadRoutine(void* s)
{const char* str = (const char*)s;cout << str << endl;return (void*)1;  }
int main()
{pthread_t p1;pthread_create(&p1, nullptr, threadRoutine, (void*)"thread 1");void *retval = nullptr;int n = pthread_join(p1, &retval);if (n != 0){cerr << "errno: " << errno << strerror(errno) <<endl;}// 当前环境是64位的，int是32位的，如果强转为32位会报错。cout << "wait successful!" << " retval: " << (int64_t)retval << endl;return 0;
}

3.4 线程分离

如果新线程不用被等待获取返回值，可以在主线程中分离该线程

• pthread_detach

• 分离指定id的线程

#include<iostream>
#include<pthread.h>
#include <unistd.h>
#include <cstring>using namespace std;void* threadRoutine(void* s)
{const char* str = (const char*)s;cout << str << endl;  
}
int main()
{pthread_t p1;pthread_create(&p1, nullptr, threadRoutine, (void*)"thread 1");pthread_detach(p1);sleep(3);return 0;
}

3.5 线程退出

线程退出有三种方式：

1. return，线程正常执行完退出
2. pthread_exit, 谁调用该函数谁退出，retval为返回值。
3. pthread_cancel：让指定id的线程退出，退出码为PTHREAD_CANCELED（-1）
   

3.6 获取线程id

• pthread_self：可以获取当前线程的id

#include<iostream>
#include<pthread.h>
#include <unistd.h>
#include <cstring>using namespace std;void* threadRoutine(void* s)
{const char* str = (const char*)s;cout << str << endl;}
int main()
{pthread_t p1;pthread_create(&p1, nullptr, threadRoutine, (void*)"thread 1");pthread_detach(p1);cout << pthread_self() << endl;sleep(3);return 0;
}

3.7 线程id与LWP的区别

pthread库是共享库，也会被映射到进程地址空间中的共享区。在pthread库中创建的所有线程会被组织成数组。而线程id就是数组中线程的地址。LWP是轻量级进程使用的，被封装进struct pthread字段中。这种结构类似文件，struct FILE中封装了文件描述符fd。

• 每一个线程都拥有自己的栈，主线程的栈是进程地址空间的栈，而其他线程的栈都在共享区中
• 所有线程共享大部分数据，如果创建了一个全局变量，但是想让每个线程都拥有这个变量，那么就需要在前面加__thread,然后每个线程都会在自己的局部存储里面创建这个全局变量。其生命周期是全局的。

#include<iostream>
#include<pthread.h>
#include <unistd.h>
#include <cstring>using namespace std;
__thread int _gval = 0;void* threadRoutine(void* s)
{const char* str = (const char*)s;cout << str << "_gval: " << _gval << " &_gval: " << &_gval << endl;  
}void* threadRoutine1(void* s)
{sleep(3);const char* str = (const char*)s;cout << str << "_gval: " << _gval << " &_gval: " << &_gval << endl;  
}
int main()
{pthread_t p1, p2;pthread_create(&p1, nullptr, threadRoutine, (void*)"thread 1");pthread_create(&p2, nullptr, threadRoutine1, (void*)"thread 2");sleep(4);return 0;
}

• 上面两个线程打印出来的地址一定不同。
  

四.线程互斥

• 临界资源：多个执行流共享的资源就是临界资源
• 临界区：访问临界资源的代码就是临界区
• 互斥：任一时刻，只有一个执行流进入临界区。
• 原子性：一个操作只有两种状态，要么还没开始，要么就已完成。

在多线程的场景下，不同线程访问临界资源会引发线程安全的问题，例如：对临界资源做自增操作。

#include<iostream>
#include<pthread.h>
#include <unistd.h>
#include <cstring>int _gval = 0;void* threadRoutine(void* s)
{	int cnt = 0;while (cnt < 100){cnt++;}
}int main()
{pthread_t p1, p2;pthread_create(&p1, nullptr, threadRoutine, nullptr);pthread_create(&p2, nullptr, threadRoutine1, nullptr);sleep(4);   return 0;
}

• cnt++，这一条代码会被解释为三条汇编指令：

如果一个线程执行到第二步的时候时间片到了，切换为另一个线程，然后再切换回第一个线程，那么第二个线程对cnt的操作就会被覆盖。如果要保证对cnt的操作是线程安全的，即没有并发访问问题，就需要对该临界区加锁，让临界区任何时刻只能有一个线程进入。

4.1 互斥锁接口

锁的接口定义在pthread.h头文件中。

1. 定义

pthread_mutex_t mutex;

• 定义mutex变量

2. 初始化

• pthread_mutex_t* mutex ：传入外部定义的锁。
• const pthread_mutexattr_t* attr：锁的属性，一般设置为nullptr

3. 销毁

• pthread_mutex_t* mutex ：传入外部定义的锁

如果定义的锁是全局变量，可以使用PTHREAD_MUTEX_INITIALIZER初始化，不需要调用pthread_mutex_init() 和 pthread_mutex_destroy()

4. 加锁

5. 解锁

6. 代码示例

#include<iostream>
#include<pthread.h>
#include <unistd.h>
#include <cstring>int _gval = 0;// 写法一；
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;void* threadRoutine(void* s)
{	int cnt = 0;pthread_mutex_lock(&mutex);while (cnt < 100){cnt++;pthread_mutex_unlock(&mutex);}}int main()
{pthread_t p1, p2;pthread_create(&p1, nullptr, threadRoutine, nullptr);pthread_create(&p2, nullptr, threadRoutine1, nullptr);sleep(4);   return 0;
}// 写法二：void* threadRoutine(void* s)
{	pthread_mutex_t* mutex = (pthread_mutex_t*)s;int cnt = 0;pthread_mutex_lock(mutex);while (cnt < 100){cnt++;pthread_mutex_unlock(mutex);}}int main()
{pthread_mutex_t mutex;pthread_mutex_init(&mutex, nullptr);pthread_t p1, p2;pthread_create(&p1, nullptr, threadRoutine, (void*)&mutex);pthread_create(&p2, nullptr, threadRoutine1, (void*)&mutex);pthread_mutex_destroy(&mutex);sleep(4);   return 0;
}

4.2 锁的原理

锁也是共享资源，访问共享资源就会有线程安全问题。为了防止这种套娃情况的发生，加锁和解锁操作具有原子性，大多数体系结构都提供了swap或exchange指令，这两条指令都保证了加锁的原子性。

加锁： 伪代码
lock：
movb $0, %al
xchgb %al, mutex ;这条指令是核心，将锁转移到当前线程的上下文中，这也就相当于线程取得了锁资源
if al > 0 then return 0
else 挂起等待
goto lock
解锁：
movb $1 ,mutex
唤醒等待mutex的线程
return 0;

在上面代码中，1不会增多，只会在不同线程之间流转，从而保证了多个线程只有一个锁资源

4.3 死锁

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

1. 互斥
2. 请求与保持：线程请求资源的过程中，不放弃持有资源
3. 不剥夺：一个执行流已获得的资源，在末使用完之前，不能强行剥夺
4. 环路等待：若干执行流之间形成一种头尾相接的循环等待资源的关系

预防死锁：破坏四个条件之一

1. 不加锁可以破坏互斥条件
2. 主动释放锁可以破坏请求与保持条件
3. 控制线程释放锁资源可以破坏不剥夺条件
4. 按照顺序申请锁可以破坏环路等待条件
   

五.线程同步

如果一个线程释放锁之后又立即申请锁，那么可能会导致其他线程一直申请不到锁，这样就会形成饥饿问题。这样明显是不合理的，我们可以使用条件变量来防止发生饥饿问题。一般条件变量是配合互斥锁使用的。条件变量的接口定义在pthread.h头文件中。

5.1 条件变量

1. 定义

pthread_cond_t

• 定义一个条件变量

2. 初始化

• pthread_cond_t* cond ：传入外部定义的条件变量。
• const pthread_condxattr_t* attr：条件变量的属性，一般设置为nullptr

3. 销毁

4. 阻塞

• pthread_cond_t* cond：阻塞cond条件变量
• pthread_mutex_t *mutex：当线程被阻塞时会释放锁，等到线程被唤醒时会重新申请锁

5. 唤醒

• 唤醒在条件变量下等待的线程有两种方式：1.唤醒一个线程 2.唤醒所有线程
• pthread_cond_singal() ：唤醒在条件变量cond下等待的一个线程
• pthread_cond_broadcast() ：唤醒在条件变量cond下等待的所有线程
  

5.2 posix信号量

  POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 POSIX可以用于线程间同步。信号量是资源的一种预定机制，将共享资源看作多份。信号量相关的接口在semaphore.h头文件中。

1. 定义信号量

sem_t

• 信号量类型，可以定义信号量

2. 初始化

• pshared:0表示线程间共享，非零表示进程间共享
• value：信号量初始值，资源的数量
• 当value=1时，为二元信号量，效果等同于互斥锁。

2. 销毁

3. 申请信号量资源

• 对信号量初始值(value)-1

4. 归还信号量资源

• 对信号量初始值(value)+1
  

六.生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

如果我们想要维护好生产者消费者模型，就要研究3种关系，两种角色，一个缓冲区

• 三种关系
  • 生产者与生产者：具有互斥关系
  • 消费者与消费者：具有互斥关系
  • 生产者与消费者：具有互斥和同步关系。
    • 同步：当缓冲区满时，让消费者消费，当缓冲区空时，让生产者生产
• 两种角色
  • 生产者
  • 消费者
• 一个缓冲区
  • 基于阻塞队列
  • 基于环形队列

生产者消费者模型的优点：

1. 效率高
   1. 效率高体现在可以并发生产任务和并发处理任务
2. 支持忙闲不均
   1. 生产者生产任务之后不必等消费者消费，而是将任务放到缓冲区，可以继续进行生产任务。
      

6.1 基于阻塞队列(BlockQueue)式的cp问题

在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)。

• 主要用到互斥锁和条件变量

生产者可以生产一系列任务让消费者执行任务。下面是模拟的一个计算器任务。

#pragma once#include <iostream>
#include <string>
#include <unordered_map>
#include <functional>class Task
{
public:Task() = default;Task(int x, int y, char op) : _x(x), _y(y), _op(1, op){ }void run(){std::unordered_map<std::string, std::function<void(void)>> cal = {{"+", [this]{ this->_result = this->_x + this->_y; }},{"-", [this]{ this->_result = this->_x - this->_y; }},{"*", [this]{ this->_result = this->_x * this->_y; }},{"/", [this]{ if (this->_y == 0) this->_exitStatus = 2; this->_result = this->_x/this->_y; }},{"%", [this]{ if (this->_y == 0) this->_exitStatus = 1; this->_x%this->_y; }}};cal[_op]();}void formatExpress(){std::cout << "productor : " << _x << " " << _op << " " << _y << " = ?" << std::endl;}void formatRes(){std::cout << "consumer : " << _x << " " << _op << " " << _y << " = " << _result << std::endl;}int getExitCode(){return _exitStatus;}int getRes(){return _result;}private:int _x;int _y;std::string _op;int _result;int _exitStatus;
};

#pragma once
#include <iostream>
#include <pthread.h>
#include <memory>
#include <vector>
#include <queue>
#include <unistd.h>using std::cout;
using std::endl;const int defaultCapacity = 5;template <class T>
class BlockQueue
{public:BlockQueue(int cap = defaultCapacity): _cap(cap){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_consumerCond, nullptr);pthread_cond_init(&_productorCond, nullptr);}bool isFull(){return _bq.size() == _cap;}void push(const T &in){// 生产者生产数据pthread_mutex_lock(&_mutex);// 如果缓冲区满，则生产者阻塞while (isFull()) pthread_cond_wait(&_productorCond, &_mutex);  _bq.push(in);/// 可以采用一定策略唤醒消费者消费资源pthread_cond_signal(&_consumerCond);pthread_mutex_unlock(&_mutex);}bool isempty(){return _bq.empty();}void pop(T* out){pthread_mutex_lock(&_mutex);while (isempty())pthread_cond_wait(&_consumerCond, &_mutex);*out = _bq.front();_bq.pop();pthread_mutex_unlock(&_mutex);pthread_cond_signal(&_productorCond);}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_consumerCond);pthread_cond_destroy(&_productorCond);}private:// 缓冲区std::queue<T> _bq;// 缓冲区容量int _cap;// 互斥锁pthread_mutex_t _mutex;// 条件变量：当缓冲区满的时候，生产者阻塞，当缓冲区空的时候，消费者阻塞pthread_cond_t _consumerCond;pthread_cond_t _productorCond;
};

#include "bQueue.hpp"
#include "Task.hpp"
#include <cstring>void *consume(void *args)
{BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);while (true){// 从缓存区拿资源Task t;bq->pop(&t);// 使用资源t.run();t.formatRes();sleep(1);}return nullptr;
}void *product(void *args)
{BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);const char* s = "+-*/%";while (true){sleep(1);// 生产资源int x = rand()%10;   int y = rand()%10;   char op = s[rand()%strlen(s)];Task t(x, y, op);t.formatExpress();  // 将资源放到缓冲区中bq->push(t);}return nullptr;
}int main()
{srand(time(nullptr));int n = 5;BlockQueue<Task> *bq = new BlockQueue<Task>(n);std::vector<pthread_t> consumers(3);std::vector<pthread_t> productors(4);for (auto &pth : consumers){pthread_create(&pth, nullptr, consume, static_cast<void *>(bq));}for (auto &pth : productors){pthread_create(&pth, nullptr, product, static_cast<void *>(bq));}for (auto tid : consumers)pthread_join(tid, nullptr);for (auto tid : productors)pthread_join(tid, nullptr);return 0;
}

6.2 基于环形队列的cp问题

下面实现一种基于环形队列的生产者消费者模型，采用数组模型环形队列。

在阻塞队列中，我们将缓冲区看作一个整体使用，因此也只定义了一个互斥锁就可以保证线程安全。但在环形队列中，我们将缓冲区看成n个小空间使用，因此使用信号量，只要消费者和生产者不指向同一个小空间，它们就可以并发执行。
消费者关心的是数据个数，生产者关心的是剩余空间个数，因此我们可以定义两个信号量，当剩余空间为0时，阻塞生产者；当数据个数为0时，阻塞消费者。
生产者与生产者之间互斥关系要靠一个互斥锁，消费者与消费者互斥关系也要靠一个互斥锁，故需要定义两个互斥锁。不能设置成一个互斥锁，因为生产者与消费者之间除非所占空间相同，不然没有互斥关系。

#pragma once#include <iostream>
#include <pthread.h>
#include <memory>
#include <vector>
#include <queue>
#include <unistd.h>
#include <semaphore.h>using std::cout;
using std::endl;
const int N = 5;template <class T>
class RingQueue
{
public:RingQueue(int cap = N) : _cap(cap), _consumerIndex(0), _productorIndex(0), _vRingQueue(cap){pthread_mutex_init(&_consumerMutex, nullptr);pthread_mutex_init(&_productorMutex, nullptr);sem_init(&_spaceSem, 0, _cap);sem_init(& _dataSem, 0, 0);}void push(const T &in){// 生产者生产数据P(_spaceSem);lock(_productorMutex);_vRingQueue[_productorIndex++] = in;_productorIndex %= _cap;unlock(_productorMutex);V(_dataSem);}void pop(T *out){P(_dataSem);lock(_consumerMutex);*out = _vRingQueue[_consumerIndex++];_consumerIndex %= _cap;unlock(_consumerMutex);V(_spaceSem);}~RingQueue(){pthread_mutex_destroy(&_consumerMutex);pthread_mutex_destroy(&_productorMutex);sem_destroy(&_spaceSem);sem_destroy(&_dataSem);}private:void lock(pthread_mutex_t &mutex){pthread_mutex_lock(&mutex);}void unlock(pthread_mutex_t &mutex) {   pthread_mutex_unlock(&mutex);}void P(sem_t &sem){  sem_wait(&sem);}void V(sem_t &sem){sem_post(&sem);}private:std::vector<T> _vRingQueue;int _cap; // 环形队列容量sem_t _spaceSem; // 生产者关系的是空间资源sem_t _dataSem;  // 消费者关心的是数据资源pthread_mutex_t _consumerMutex;pthread_mutex_t _productorMutex;// 消费者访问空间int _consumerIndex;// 生产者访问空间int _productorIndex;
};

七.常见概念

1. STL中的容器不是线程安全的
2. 智能指针是线程安全的
   1. unique_ptr：不涉及全局变量，只在局部有效
   2. shared_ptr：引用计数有线程安全问题，但是设计者采用了CAS操作，保证了线程安全
3. 悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
4. 乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
5. 自旋锁：非阻塞的申请锁资源。如果访问临界区的时间短可以采用自旋锁。
   1. pthread_spinlock_t：可以定义自旋锁