C++11中的多線(xiàn)程的支持

千禧年以后，主流的芯片廠(chǎng)商都開(kāi)始生產(chǎn)多核處理器，所以并行編程越來(lái)越重要了。在C++98中根本沒(méi)有自己的一套多線(xiàn)程編程庫(kù),它采用的是C99中的POSIX標(biāo)準(zhǔn)的pthread庫(kù)中的互斥鎖,來(lái)完成多線(xiàn)程編程。

首先來(lái)簡(jiǎn)單一個(gè)概念:原子操作,即多線(xiàn)程程序中"最小的且不可以并行化的操作"。通俗來(lái)說(shuō),如果對(duì)一個(gè)資源的操作是原子操作,就意味著一次性只有一個(gè)線(xiàn)程的一個(gè)原子操作可以對(duì)這個(gè)資源進(jìn)行操作。在C99中,我們一般都是采用互斥鎖來(lái)完成粗粒度的原子操作。

#include<pthread.h>
#include<iostream>
using namespace std;

static long long total =0;
pthread_mutex_t m=PTHREAD_MUTEX_INITIALIZER;//互斥鎖

void * func(void *)
{
    long long i;
    for(i=0;i<100000000LL;i++)
    {
        pthread_mutex_lock(&m);
        total +=i;
        pthread_mutex_unlock(&m);
    }
}

int main()
{
    pthread_t thread1,thread2;
    if(pthread_create(&thread1,nullptr,&func,nullptr))
    {
        throw;
    }
    if(pthread_create(&thread2,nullptr,&func,nullptr))
    {
        throw;
    }
    pthread_join(thread1,nullptr);
    pthread_join(thread2,nullptr);
    cout<<total<<endl;//9999999900000000
}

可以看出來(lái),上書(shū)代碼中total +=i;就是原子操作。

1.C++11中的原子類(lèi)型

我們發(fā)現(xiàn),在C99中的互斥鎖需要顯式聲明,要自己開(kāi)關(guān)鎖,為了簡(jiǎn)化代碼,C++11中定義了原子類(lèi)型。這些原子類(lèi)型是一個(gè)class,它們的接口都是原子操作。如下所示:

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

atomic_llong total {0};//原子數(shù)據(jù)類(lèi)型

void func(int)
{
    for(long long i=0;i<100000000LL;i++)
    {
        total+=i;
    }
}
int main()
{
    thread t1(func,0);
    thread t2(func,0);

    t1.join();
    t2.join();
    cout<<total<<endl;//9999999900000000
}

上述代碼中total就是一個(gè)原子類(lèi)對(duì)象,它的接口例如這里的重載operator+=()就是一個(gè)原子操作,所以我們不需要顯式調(diào)用互斥鎖了。

總共有多少原子類(lèi)型呢?C++11的做法是,存在一個(gè)atomic類(lèi)模板,我們可以通過(guò)這個(gè)類(lèi)模板定義出想要的原子類(lèi)型:

using atomic_llong = atomic<long long>;

所以我們想把什么類(lèi)型搞成原子類(lèi)型,只需要傳入不同的模板實(shí)參就行了。

總之,C++11中原子操作就是atomic模板類(lèi)的成員函數(shù)。

1.1 原子類(lèi)型的接口

我們知道原子類(lèi)型的接口就是原子操作,但是我們現(xiàn)在關(guān)注一下,它們有哪些接口?

原子類(lèi)型屬于資源類(lèi)數(shù)據(jù),多個(gè)線(xiàn)程只能訪(fǎng)問(wèn)單個(gè)預(yù)祝你類(lèi)型的拷貝。所以C++11中的原子類(lèi)型不支持移動(dòng)語(yǔ)義和拷貝語(yǔ)義,原子類(lèi)型的操作都是對(duì)那個(gè)唯一的一份資源操作的,原子類(lèi)型沒(méi)有拷貝構(gòu)造,拷貝賦值,移動(dòng)構(gòu)造和移動(dòng)賦值的。

atomic<float> af{1.2f};//正確
atomic<float> af1{af};//錯(cuò)誤,原子類(lèi)型不支持拷貝語(yǔ)義
float f=af;//正確,調(diào)用了原子類(lèi)型的接口
af=0.0;//正確,調(diào)用了原子類(lèi)型的接口

看一下上表中的一些原子類(lèi)型的接口,load()是進(jìn)行讀取操作的,例如

atomic<int> a(2);
int b=a;
b=a.load();

上如代碼中的,b=a就是等價(jià)于b=a.load(),實(shí)際上,atomic<int>中存在operator int()接口,這個(gè)接口中:

operator __int_type() const noexcept
{ return load(); }

store()接口是用來(lái)寫(xiě)數(shù)據(jù)的的:

atomic<int> a;
a=1;
a.store(1);

上述代碼中a=1相當(dāng)于a.load(1),atomic<int>中存在operator=(int)接口,它的實(shí)現(xiàn)如下:

__int_type operator=(__int_type __i) noexcept
{
store(__i);
return __i;
}

例如其他操作,比如exchange是做交換,compare_exchange_weak/strong()是比較并交換(CAS操作的),它們的實(shí)現(xiàn)會(huì)更復(fù)雜一些,還有一些符號(hào)的重載,這里就不一一介紹了,<<C++ Concurrency in Action>>的第5章和第7章會(huì)詳細(xì)介紹這部分內(nèi)容。

值得注意的是,這里有一個(gè)特殊的原子類(lèi)型atomic_flag,這是一個(gè)原子類(lèi)型是無(wú)鎖的,也就是說(shuō)線(xiàn)程對(duì)這種類(lèi)型的數(shù)據(jù)的訪(fǎng)問(wèn)是無(wú)鎖的,所以它就不需要接口:load和store,即多個(gè)線(xiàn)程可以同時(shí)操作這個(gè)資源。我們可以用它來(lái)實(shí)現(xiàn)自旋鎖

1.2簡(jiǎn)單自旋鎖的實(shí)現(xiàn)

互斥鎖是說(shuō),當(dāng)一個(gè)線(xiàn)程訪(fǎng)問(wèn)一個(gè)資源的時(shí)候,他會(huì)給進(jìn)入臨界區(qū)代碼設(shè)置一把鎖,出來(lái)的時(shí)候就把鎖給打開(kāi),當(dāng)其他進(jìn)程要進(jìn)入臨界區(qū)的時(shí)候,就會(huì)實(shí)現(xiàn)看一下鎖,如果鎖是關(guān)的,那就阻塞自己,這樣core就會(huì)去執(zhí)行其他工作,導(dǎo)致上下文切換嗎,所以它的效率會(huì)比較低。原子類(lèi)型中is_lock_free()就是說(shuō)明的這個(gè)原子類(lèi)型的訪(fǎng)問(wèn)是否使用的互斥鎖。

與互斥鎖相反的是自旋鎖,區(qū)別是,當(dāng)其他進(jìn)程要進(jìn)入臨界區(qū)的時(shí)候,如果鎖是關(guān)的,它不會(huì)阻塞自己,而是不斷查看鎖是不是開(kāi)的,這樣就不會(huì)引發(fā)上下文切換,但是同樣也會(huì)增加cpu利用率。

我們可以使用atomic_flag原子類(lèi)型來(lái)實(shí)現(xiàn)自旋鎖,因?yàn)樵赼tomic_flag本身是無(wú)鎖的,所以多個(gè)線(xiàn)程可以同時(shí)訪(fǎng)問(wèn)它,相當(dāng)于同時(shí)訪(fǎng)問(wèn)這把自旋鎖,實(shí)現(xiàn)如下:

#include<thread>
#include<atomic>
#include<iostream>
#include<unistd.h>
using namespace std;
atomic_flag lock=ATOMIC_FLAG_INIT;//獲得自旋鎖
void f(int n)
{
    while(lock.test_and_set())
    {//嘗試獲得原子鎖
       cout<<"Waiting from thread "<<n<<endl;
    }
    cout<<"Thread "<<n<<" starts working"<<endl;
}
void g(int n)
{
    cout<<"Thread "<<n<<" is going to start"<<endl;
    lock.clear();//打開(kāi)鎖
    cout<<"Thread "<<n<<" starts working"<<endl;
}
int main()
{
    lock.test_and_set();//關(guān)上鎖
    thread t1(f,1);
    thread t2(g,2);
    t1.join();
    usleep(100000);
    t2.join();
}

這里的test_and_set()是一個(gè)原子操作,它做的是,寫(xiě)入新值并返回舊值。在main()中,我們首先給這個(gè)lock變量,寫(xiě)入true值,即關(guān)上鎖,然后再線(xiàn)程t1中,它不斷嘗試獲得自旋鎖,再線(xiàn)程t2中,clear()接口,相當(dāng)于將lock變量值變成false,這時(shí)自旋鎖就打開(kāi)了,這樣子,線(xiàn)程t1就可以執(zhí)行剩下的代碼了。

簡(jiǎn)單的我們可以將lock封裝一下

void Lock(atomic_flag & lock)
{
    while(lock.test_and_set());
}
void Unlock(atomic_flag & lock)
{
    lock.clear();
}

上面操作中,我們就相當(dāng)于完成了一把鎖,可以用其實(shí)現(xiàn)互斥訪(fǎng)問(wèn)臨界區(qū)的功能了。不過(guò)這個(gè)和C99中的pthread_mutex_lock()和pthread_mutex_unlock()不一樣,C99中的這兩個(gè)鎖是互斥鎖,而上面代碼中實(shí)現(xiàn)的是自旋鎖。

2.提高并行程度

#include <thread>
#include <atomic>
 
atomic<int> a;
atomic<int> b;
void threadHandle()
{
     int t = 1;
     a = t;
     b = 2; // b 的賦值不依賴(lài) a
}

在上面代碼中,對(duì)a和b的賦值語(yǔ)句實(shí)際上可以不管先后的,如果允許編譯器或者硬件對(duì)其重排序或者并發(fā)執(zhí)行,那就會(huì)提高并行程度。

在單線(xiàn)程程序中,我們根部不關(guān)心它們的執(zhí)行順序,反正結(jié)果都是一樣的,但是多線(xiàn)程不一樣,如果執(zhí)行順序不一樣,結(jié)果就會(huì)不同。

#include <thread>
#include <atomic>
#include<iostream>
using namespace std;
atomic<int> a{0};
atomic<int> b{0};
void ValueSet(int )
{
     int t = 1;
     a = t;
     b = 2; // b 的賦值不依賴(lài) a
}
int Observer(int)
{
    cout<<"("<<a<<","<<b<<")"<<endl;
}
int main()
{
    thread t1(ValueSet,0);
    thread t2(Observer,0);
    t1.join();
    t2.join();
    cout<<"Final: ("<<a<<","<<b<<")"<<endl;
}

上面代碼中,Observer()中的輸出結(jié)果,會(huì)和a和b的賦值順序有關(guān),它的輸出結(jié)果肯可能是:(0,0) (1,0) (0,2) (1,2)。這就說(shuō)明了,多線(xiàn)程程序中,如果執(zhí)指令行順序不一樣,結(jié)果就會(huì)不同。

影響并行程度的兩個(gè)關(guān)鍵因素是:編譯器是否有權(quán)對(duì)指令進(jìn)行重排序和硬件是否有權(quán)對(duì)匯編代碼重排序。

C++11中,我們可以顯式得告訴編譯器和硬件它們的權(quán)限,進(jìn)而提高并發(fā)程度。通俗來(lái)說(shuō),如果我們要求并行程度最高,那么我們就授權(quán)給編譯器和硬件,允許它們重排序指令。

2.1 memory_order的參數(shù)

原子類(lèi)型的成員函數(shù)中,大多數(shù)都可以接收一個(gè)類(lèi)型為memory_order的參數(shù),它就是可以告訴編譯器和硬件,是否可以重排序。

typedef enum memory_order {
    memory_order_relaxed,    // 不對(duì)執(zhí)行順序做保證
    memory_order_acquire,    // 本線(xiàn)程中,所有后續(xù)的讀操作必須在本條原子操作完成后執(zhí)行
    memory_order_release,    // 本線(xiàn)程中,所有之前的寫(xiě)操作完成后才能執(zhí)行本條原子操作
    memory_order_acq_rel,    // 同時(shí)包含 memory_order_acquire 和 memory_order_release
    memory_order_consume,    // 本線(xiàn)程中,所有后續(xù)的有關(guān)本原子類(lèi)型的操作,必須在本條原子操作完成之后執(zhí)行
    memory_order_seq_cst    // 全部存取都按順序執(zhí)行
    } memory_order;

在C++11中,memory_order的參數(shù)的默認(rèn)值是memory_order_seq_cst,即不允許編譯器和硬件進(jìn)行重排序,這樣一來(lái),在上嗎代碼中的Observer()中輸出結(jié)果就不可能是(0,2),因?yàn)閷?duì)a的賦值語(yǔ)句是先于b的。這實(shí)際上就是:順序一致性,準(zhǔn)確來(lái)說(shuō)就是在同一個(gè)線(xiàn)程中,原子操作的順序和代碼的順序保持一致。

而如果我們改動(dòng)一下代碼:

#include <thread>
#include <atomic>
#include<iostream>
using namespace std;
atomic<int> a{0};
atomic<int> b{0};
void ValueSet(int )
{
     int t = 1;
     a.store(t,memory_order_relaxed);
     b.store(2,memory_order_relaxed); // b 的賦值不依賴(lài) a
}
int Observer(int)
{
    cout<<"("<<a<<","<<b<<")"<<endl;
}
int main()
{
    thread t1(ValueSet,0);
    thread t2(Observer,0);
    t1.join();
    t2.join();
    
    
    cout<<"Final: ("<<a<<","<<b<<")"<<endl;
}

在上面代碼中的Observer()中輸出結(jié)果是有可能是:(0,2)的,因?yàn)檫@里的memory_order_relaxed不對(duì)原子操作的順序有嚴(yán)格要求,就有可能發(fā)生b先被賦值了,而此時(shí)a還沒(méi)被賦值的情況。

所以,為了進(jìn)一步開(kāi)發(fā)原子操作的并行程度,我們的目標(biāo)是:保證程序既快又對(duì)。

2.2 release-acquire內(nèi)存順序

#include <thread>
#include <atomic>
#include<iostream>
using namespace std;
atomic<int> a;
atomic<int> b;
void Thread1(int )
{
     int t = 1;
     a.store(t,memory_order_relaxed);
     b.store(2,memory_order_release); // 本操作前的寫(xiě)操作必須先完成,即保證a的賦值快于b
}
void Thread2(int )
{
    while(b.load(memory_order_acquire)!=2);//必須等該原子操作完成后,才執(zhí)行下面代碼
    cout<<a.load(memory_order_relaxed)<<endl;//1

}
int main()
{
    thread t1(Thread1,0);
    thread t2(Thread2,0);
    t2.join();
    t1.join();
}

上面代碼中,實(shí)際上也是實(shí)現(xiàn)了一種自旋鎖的操作,我們保證了a.store快于b.store,而b.load又一定快于a.load。而且,對(duì)于b的store和load就實(shí)現(xiàn)了一種release-acquire內(nèi)存順序.

2.3 release-consume內(nèi)存順序

#include<thread>
#include<atomic>
#include<cassert>
#include<string>
using namespace std;

atomic<string*> ptr;
atomic<int> date;
void Producer()
{
    string *p=new string("hello");
    date.store(42,memory_order_relaxed);
    ptr.store(p,memory_order_release);//date賦值快于ptr
}
void Consumer()
{
    string *p2;
    while(!(p2=ptr.load(memory_order_consume)));
    assert(*p2=="hello");//一定成立
    assert(date.load(memory_order_relaxed)==42);//可能斷言失敗,因?yàn)檫@個(gè)指令可能在本線(xiàn)程中首先執(zhí)行
}
int main()
{
    thread t1(Producer);
    thread t2(Consumer);
    t1.join();
    t2.join();
}

上面的內(nèi)存順序也叫生產(chǎn)者-消費(fèi)者順序。

實(shí)際上,總共的內(nèi)存模型就是4個(gè):順序一致性,松散的(relaxed),release-consume和release-acquire。

2.4 小結(jié)

實(shí)際上,對(duì)于并行編程來(lái)說(shuō),最根本的的在于,并行算法,而不是從硬件上搞內(nèi)存模型優(yōu)化啥的,如果你嫌麻煩的話(huà),全部使用順序一致性?xún)?nèi)存模型,對(duì)并行效率的影響也不是很大。

3.線(xiàn)程局部存儲(chǔ)

線(xiàn)程擁有自己的?？臻g,但是堆空間,靜態(tài)數(shù)據(jù)區(qū)(文件data,bss段,全局/靜態(tài)變量)是共享的。線(xiàn)程之間互相共享,靜態(tài)數(shù)據(jù)當(dāng)然是很好的,但是我們也需要線(xiàn)程自己的局部變量

#include<pthread.h>
#include<iostream>
using namespace std;

int thread_local errorCode=0;
void* MaySetErr(void *input)
{
    if(*(int*)input==1)
        errorCode=1;
    else if(*(int*)input==2)
        errorCode=2;
    else
        errorCode=0;
    cout<<errorCode<<endl;
}
int main()
{
    int input_a=1;
    int input_b=2;
    pthread_t thread1,thread2;
    pthread_create(&thread1,nullptr,&MaySetErr,&input_a);
    pthread_create(&thread2,nullptr,&MaySetErr,&input_b);
    pthread_join(thread1,nullptr);
    pthread_join(thread2,nullptr);
    cout<<errorCode<<endl;//0
}

上面代碼中的errorCode是一個(gè)thread_local變量,它意味著它是一個(gè)線(xiàn)程內(nèi)部的全局變量,線(xiàn)程開(kāi)始時(shí),他會(huì)被初始化,然后線(xiàn)程結(jié)束時(shí),該值就不會(huì)有效。實(shí)際上兩個(gè)進(jìn)程中會(huì)有各自的errorCode,而main函數(shù)中也有自己的errorCode

4.快速退出

在C++98中,我們會(huì)見(jiàn)到3中終止函數(shù):terminate,abort,exit。而在C++11中我們?cè)黾恿藂uick_exit終止函數(shù),這種終止函數(shù)主要用在線(xiàn)程種。

1.terminate函數(shù),它是C++種的異常機(jī)制有關(guān)的,通常沒(méi)有被捕獲的異常就會(huì)調(diào)用terminate

2.abort函數(shù)是底層的終止函數(shù),terminate就是調(diào)用它來(lái)終止進(jìn)程的,但是abort調(diào)用時(shí),不會(huì)調(diào)用任何析構(gòu)函數(shù),會(huì)引發(fā)內(nèi)存泄漏啥的,但是一般來(lái)說(shuō),他會(huì)給符合POSIX的操作系統(tǒng)拋出一個(gè)信號(hào),此時(shí)signal handler就會(huì)默認(rèn)的釋放進(jìn)程種的所有資源,來(lái)避免內(nèi)存泄漏。

3.exit函數(shù)是正常退出,他會(huì)調(diào)用析構(gòu)函數(shù),但是有時(shí)候析構(gòu)函數(shù)狠復(fù)雜,那我們還不如直接調(diào)用absort函數(shù),將釋放資源的事情留給操作系統(tǒng)。

在多線(xiàn)程情況下,我們一般都是采用exit來(lái)退出的,但是這樣容易卡死,當(dāng)線(xiàn)程復(fù)雜的時(shí)候,exit這種正常退出方式,太過(guò)于保守了,但是abort這種退出方式又太激進(jìn)了,所以有一種新的退出函數(shù):quick_exit函數(shù)。

quick_exit

這個(gè)函數(shù)不執(zhí)行析構(gòu)函數(shù),而使得程序終止,但是和abort不同的是,abort一般都是異常退出,而quick_exit是正常退出。

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

struct A{~A(){cout<<"Destruct A."<<endl;}};
void closeDevice(){cout<<"device is closed."<<endl;}
int main()
{
    A a;
    at_quick_exit(closeDevice);
    quick_exit(0);
}
//樣容易卡死,當(dāng)線(xiàn)程復(fù)雜的時(shí)候,`exit`這種正常退出方式,太過(guò)于保守了,但是`abort`這種退出方式又太激進(jìn)了,所以有一種新的退出函數(shù):`quick_exit`函數(shù)。

//這個(gè)函數(shù)不執(zhí)行析構(gòu)函數(shù),而使得程序終止,但是和`abort`不同的是,`abort`一般都是異常退出,而`quick_exit`是正常退出。
#include<cstdlib>
#include<iostream>
using namespace std;

struct A{~A(){cout<<"Destruct A."<<endl;}};
void closeDevice(){cout<<"device is closed."<<endl;}
int main()
{
    A a;
    at_quick_exit(closeDevice);
    quick_exit(0);
}

以上就是C++11學(xué)習(xí)之多線(xiàn)程的支持詳解的詳細(xì)內(nèi)容，更多關(guān)于C++11多線(xiàn)程的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論