异构计算关键技术之多线程技术（二）

异构计算关键技术之多线程技术（二）

诞生伊始，计算机处理能力就处于高速发展中。及至最近十年，随着大数据、区块链、AI 等新技术的持续火爆，人们为提升计算处理速度更是发展了多种不同的技术思路。大数据受惠于分布式集群技术，区块链带来了专用处理器（Application-Specific IC, ASIC）的春天，AI 则让大众听到了“异构计算”这个计算机界的学术名词。

“异构计算”（Heterogeneous computing），是指在系统中使用不同体系结构的处理器的联合计算方式。在 AI 领域，常见的处理器包括：CPU（X86，Arm，RISC-V 等），GPU，FPGA 和 ASIC。（按照通用性从高到低排序）。

任务分解通常包括系统分解、流程分解、任务分解等，而线程/进程技术是必不可少的。

本系列文章将介绍异构计算涉及到的内存管理技术、DMA技术、线程技术等。结合实例代码进行详细讲解多线程、DMA scatter-gather list、PCIe TLP等核心技术。

本章将介绍核心的基本概念：主要包括用户态的线程、进程技术。

一、什么是进程、线程

在C++/C学习过程中，要想“更上一层楼”的话，多线程编程是必不可少的一步。

所以，我们在看这篇文章的时候，大家需要更多的思考是为什么这么做？这样做的好处是什么？以及多线程编程都可以应用在哪里？

1. 线程与进程

进程是正在运行的程序的实例，而线程是是进程中的实际运作单位。
一个程序有且只有一个进程，但是可以拥有至少一个线程。
不同进程拥有不同的地址空间（3G虚拟地址，参考上一章的虚拟地址空间内容），互不干扰，而不同线程拥有相同的地址空间。

1705473313798.png

2. 多进程

使用多进程并发是将一个应用程序划分为多个独立的进程（每个进程只有一个线程），这些独立的进程间可以互相通信，共同完成任务。

由于操作系统对进程提供了大量的保护机制，以避免一个进程修改了另一个进程的数据，使用多进程比使用多线程更容易写出相对安全的代码。但是这也造就了多进程并发的两个缺点：

• 在进程间的通信，无论是使用信号、套接字，还是文件、管道等方式，其使用要么比较复杂，要么就是速度较慢或者两者兼而有之。
• 运行多个线程的开销很大，操作系统要分配很多的资源来对这些进程进行管理。

当多个进程并发完成同一个任务时，不可避免的是：操作同一个数据和进程间的相互通信，上述的两个缺点也就决定了多进程的并发并不是一个好的选择。所以就引入了多线程的并发。

3. 多线程

传统的C++（C++11标准之前）中并没有引入线程这个概念，在C++11出来之前，如果我们想要在C++中实现多线程，需要借助操作系统平台提供的API，比如Linux的，或者windows下的。

C++11提供了语言层面上的多线程，包含在头文件中。

它解决了跨平台的问题，提供了管理线程、保护共享数据、线程间同步操作、原子操作等类。

多线程并发指的是在同一个进程中执行多个线程。

优点： - 有操作系统相关知识的应该知道，线程是轻量级的进程，每个线程可以独立的运行不同的指令序列，但是线程不独立的拥有资源，依赖于创建它的进程而存在。 - 同一进程中的多个线程共享相同的地址空间，可以访问进程中的大部分数据，指针和引用可以在线程间进行传递。 - 同一进程内的多个线程能够很方便的进行数据共享以及通信，也就比进程更适用于并发操作。

缺点： - 由于缺少操作系统提供的保护机制，在多线程共享数据及通信时，就需要程序员做更多的工作以保证对共享数据段的操作是以预想的操作顺序进行的，并且要极力的避免死锁(deadlock)。

二、std::thread

构造&析构函数

| 函数 | 类别 | 作用 |
| thread() noexcept | 默认构造函数 | 创建一个线程，什么也不做 |
| explicit thread(Fn&& fn, Args&&… args) | 初始化构造函数 | 创建一个线程，传入参数，执行fn |
| thread(thread&& x) noexcept | 移动构造函数 | 构造一个与x相同的对象，会破坏x对象 |
| ~thread | 析构函数 | 析构对象 |

成员函数 | 函数 | 类别 | |:--------:| :---------:| | void join() | 等待线程结束并清理资源(会阻塞) | | bool joinable() | 返回线程是否可以join | | void detach() | 将线程与调用其的线程分离，彼此独立执行 | | std::thread::id get_id() | 获取线程id | | thread& operator=(thread &&rhs) | 见移动构造函数（如果对象是joinable的，那么会调用std::terminate()结果程序）|

实例：

#include <iostream>
#include <thread>
using namespace std;

void count(int id, unsigned int n)
{
    for (unsigned int i = 0; i < n; i++)
        cout << "thread " <<id<<"finish!" <<endl;
}

// 这里的线程使用了C++17
int main()
{
    thread th[10];
    for (int i = 0; i < 10; i++)
        th[i] = thread(count, i, 1000);
    for (int i = 0; i < 10; i++) 
        if (th[i].joinable())
            th[i].join();
    return 0;
}

结果：

thread 6finish!
thread 6finish!
thread 6finish!
thread 6finish!
thread 6finish!
thread 6finish!
thread 6finish!
thread 6finish!
3finish!5
thread 3finish!
thread 3finish!
thread 3finish!
thread 3finish!
thread 3finish!finish!
thread 3finish!

thread 5finish!
thread thread 3finish!
thread 3finish!
thread 3finish!
thread 3finish!
thread 3finish!
thread 3finish!

并且每次结果都不一样！！！

这是为什么呢？这就是多线程的特色！

多线程运行时是以异步方式执行的，与我们平时写的同步方式不同。异步方式可以同时执行多条语句。

总结： - 线程是在thread对象被定义的时候开始执行的，而不是在调用join函数时才执行的，调用join函数只是阻塞等待线程结束并回收资源。 - 分离的线程（执行过detach的线程）会在调用它的线程结束或自己结束时释放资源 - 线程会在函数运行完毕后自动释放，不推荐利用其他方法强制结束线程，可能会因资源未释放而导致内存泄漏。 - 没有执行join或detach的线程在程序结束时会引发异常

三、join()和detach()

每一个程序至少拥有一个线程，那就是执行main()函数的主线程。而多线程则是出现两个或两个以上的线程并行运行，即主线程和子线程在同一时间段同时运行。而在这个过程中会出现几种情况：

-   主线程先运行结束
-   子线程先运行结束
-   主子线程同时结束

在一些情况下需要在子线程结束后主线程才能结束，而一些情况则不需要等待，但需注意一点，并不是主线程结束了其他子线程就立即停止，其他子线程会进入后台运行。

1. join()

join()函数是一个等待线程完成的函数，主线程需要等待子线程结束后才可以结束。

#include <iostream>
#include <thread>
using namespace std;

void count()
{
    for (unsigned int i = 0; i < 10; i++)
        cout << "count: " << i <<endl;
}

// 这里的线程使用了C++17
//main()主线程
int main()
{
    cout << "main()"<<endl;
    cout << "main()"<<endl;
    cout << "main()"<<endl;

    thread th;
    th = thread(count);
    th.join();
    return 0;
}

结果：

main()
main()
main()
count: 0
count: 1
count: 2
count: 3
count: 4
count: 5
count: 6
count: 7
count: 8
count: 9
#include <iostream>
#include <thread>
using namespace std;

void count()
{
    for (unsigned int i = 0; i < 10; i++)
        cout << "count: " << i <<endl;
}

// 这里的线程使用了C++17
//main()主线程
int main()
{
    thread th;
    th = thread(count);
    cout << "main()"<<endl;
    cout << "main()"<<endl;
    cout << "main()"<<endl;

    th.join();
    return 0;
}

结果：

main()count: 0
count: 1
count: 2
count: 3
count: 4
count: 5
count: 6
count: 7
count: 8
count: 9

main()
main()
#include <iostream>
#include <thread>
using namespace std;

void count()
{
    for (unsigned int i = 0; i < 10; i++)
        cout << "count: " << i <<endl;
}

// 这里的线程使用了C++17
//main()主线程
int main()
{
    thread th;
    th = thread(count);
    th.join();
    cout << "main()"<<endl;
    cout << "main()"<<endl;
    cout << "main()"<<endl;

    return 0;
}

结果：

count: 0
count: 1
count: 2
count: 3
count: 4
count: 5
count: 6
count: 7
count: 8
count: 9
main()
main()
main()

2. detach()

称为分离线程函数，使用detach()函数会让线程在后台运行，即说明主线程不会等待子线程运行结束才结束。

通常称分离线程为守护线程(daemon threads),UNIX中守护线程是指，没有任何显式的用户接口，并在后台运行的线程。这种线程的特点就是长时间运行；线程的生命周期可能会从某一个应用起始到结束，可能会在后台监视文件系统，还有可能对缓存进行清理，亦或对数据结构进行优化。

#include <iostream>
#include <thread>
using namespace std;

void count()
{
    for (unsigned int i = 0; i < 10; i++)
        cout << "count: " << i <<endl;
}

// 这里的线程使用了C++17
//main()主线程
int main()
{
    cout << "main()"<<endl;
    cout << "main()"<<endl;
    cout << "main()"<<endl;
    thread th;
    th = thread(count);
    th.detach();

    return 0;
}

1705548646607.png

可以看到，子线程还没有执行完，主线程就结束了。

总结：

join()函数：是一个等待线程的函数，主线程需等待子线程运行结束后才可以结束（注意不是才可以运行，运行是并行的）。如果打算等待对应线程，则需要细心挑选调用join()的位置。
detach()函数:子线程的分离函数，当调用该函数后，线程就被分离到后台运行，主线程不需要等待该线程结束才结束。

四、std::atomic和std::mutex

我们现在已经知道如何在c++11中创建线程，那么如果多个线程需要操作同一个变量呢？

1. 为什么要有atomic和mutex

#include <iostream>
#include <thread>
using namespace std;

int n = 0;
void count()
{
    for (unsigned int i = 0; i < 1000; i++)
        n++;
}

// 这里的线程使用了C++17
//main()主线程
int main()
{
    thread th[1000];
    for (int i = 0; i < 1000; i++)
        th[i] = thread(count);
    for (int i = 0; i < 1000; i++)
        th[i].join();

    cout << n <<endl;

    return 0;
}

几次的输出结果：

1705555056474.png

我们的输出结果应该是1000000，可是为什么实际输出结果比1000000小呢？

在上文我们分析过多线程的执行顺序——同时进行、无次序，所以这样就会导致一个问题：多个线程进行时，如果它们同时操作同一个变量，那么肯定会出错。

为了应对这种情况，c++11中出现了std::atomic和std::mutex。

2. std::mutex

std::mutex是 C++11 中最基本的互斥量，一个线程将mutex锁住时，其它的线程就不能操作mutex，直到这个线程将mutex解锁。

#include <iostream>
#include <thread>
#include <mutex>
using namespace std;

int n = 0;
mutex mtx;
void count()
{
    for (unsigned int i = 0; i < 1000; i++) {
        mtx.lock();
        n++;
        mtx.unlock();
    }
}

// 这里的线程使用了C++17
//main()主线程
int main()
{
    thread th[1000];
    for (int i = 0; i < 1000; i++)
        th[i] = thread(count);
    for (int i = 0; i < 1000; i++)
        th[i].join();

    cout << n <<endl;

    return 0;
}

1705555302811.png

| 函数 | 作用 |
| void lock() | 将mutex上锁。如果mutex已经被其它线程上锁， 那么会阻塞，直到解锁；如果mutex已经被同一个线程锁住，那么会产生死锁。 |
| void unlock() | 解锁mutex，释放其所有权。 |
| bool try_lock() | 尝试将mutex上锁。如果mutex未被上锁，则将其上锁并返回true；如果mutex已被锁则返回false。 |

2. std::atomic

mutex很好地解决了多线程资源争抢的问题，但它也有缺点：太!慢!了!

我们定义了100个thread，每个thread要循环10000次，每次循环都要加锁、解锁，这样固然会浪费很多的时间，那么该怎么办呢？接下来就是atomic大展拳脚的时间了。

#include <iostream>
#include <thread>
#include <atomic>
using namespace std;

atomic_int n = 0;
void count()
{
    for (unsigned int i = 0; i < 1000; i++) {
        n++;
    }
}

// 这里的线程使用了C++17
//main()主线程
int main()
{
    thread th[1000];
    for (int i = 0; i < 1000; i++)
        th[i] = thread(count);
    for (int i = 0; i < 1000; i++)
        th[i].join();

    cout << n <<endl;

    return 0;
}

1705555817562.png

原子操作是最小的且不可并行化的操作。

这就意味着即使是多线程，也要像同步进行一样同步操作atomic对象，从而省去了mutex上锁、解锁的时间消耗。

| 函数 | 作用 |
| atomic() noexcept = default | 构造一个atomic对象（未初始化，可通过atomic_init进行初始化） |
| constexpr atomic(T val) noexcept | 构造一个atomic对象，用val的值来初始化 |

五、std::this_thread

上面讲了那么多关于创建、控制线程的方法，现在该讲讲关于线程控制自己的方法了。

在头文件中，不仅有std::thread这个类，而且还有一个std::this_thread命名空间，它可以很方便地让线程对自己进行控制。

std::this_thread常用函数

std::this_thread是个命名空间，所以你可以使用using namespace std::this_thread;这样的语句来展开这个命名空间。

| 函数 | 作用 |
| std::thread::id get_id() noexcept | 获取当前线程id |
| template | |
| void sleep_for( const std::chrono::duration& sleep_duration ) | 等待sleep_duration（sleep_duration是一段时间） |
| void yield() noexcept | 暂时放弃线程的执行，将主动权交给其他线程（放心，主动权还会回来） |

#include <iostream>
#include <thread>
#include <atomic>
using namespace std;

atomic_bool ready = 0;
void sleep(uintmax_t ms)
{
    this_thread::sleep_for(chrono::microseconds(ms));
}

void count()
{
    while(!ready)
        this_thread::yield();

    for (int i = 0; i <= 20'0000'0000; i++);

    cout << "thread " << this_thread::get_id() <<" finished!" <<endl;
    return;
}
// 这里的线程使用了C++17
//main()主线程
int main()
{
    thread th[10];
    for (int i = 0; i < 10; i++)
        th[i] = thread(::count);
    sleep(5000);
    ready = true;
    cout << "Start!" <<endl;
    for (int i = 0; i < 10; i++)
        th[i].join();
    return 0;
}

1705574217257.png

你的输出几乎不可能和我一样，不仅是多线程并行的问题，而且每个线程的id也可能不同。

六、未完待续

下章将继续介绍核心的基本概念：内核态的线程/进程技术。

欢迎关注知乎：北京不北，+vbeijing_bubei

欢迎关注douyin：near.X （北京不北）

欢迎+V：beijing_bubei

获得免费答疑，长期技术交流。

七、参考文献

https://zhuanlan.zhihu.com/p/613630658

https://blog.csdn.net/sjc_0910/article/details/118861539