1. 程式人生 > >C++筆記之多執行緒的理解與應用

C++筆記之多執行緒的理解與應用

一、執行緒與程序

程序,擁有資源並且獨立執行的基本單位;

將CPU比作是是一個工廠,那麼程序可以看做是一個可以獨立進行工作的車間,車間內有生產必須的資源以及工人,同時工廠內同一時刻只有一個車間在開工,但是工廠內是可以有多個車間的。[1]

執行緒,程式執行的最小單元;

執行緒則是車間內的工人,工人的數量可以有多個,某些工具或者空間只有一個,需要多人分享(共享資源),如果有一個人正在使用,那麼其他工人則必須等待此人使用完畢。

程序擁有獨立的執行環境,每個程序都有自己的記憶體空間。程序間的通訊可以使用管道或者socket等。在一個程序中,執行緒共享該程序的資源,比如記憶體或者開啟的檔案。由於程序之間是隔離的,而執行緒之間共享資源,所以執行緒之間存在競爭。

二、執行緒的狀態

1、就緒

執行緒具備執行的所有條件,等待處理;

2、執行

執行緒佔了CPU資源,正在執行

3、阻塞

執行緒等待一個事件或者訊號量,此時執行緒無法執行

三、併發與並行

併發是同一時間應對多件事情的能力;而並行則是同一時間做多件事情的能力。

並行就是在相同的時間,多個工作執行單位同時執行;

在單核CPU上,多執行緒本質上是單個 CPU 的時間分片,一個時間片執行一個執行緒的程式碼,它可以支援併發處理,但是不能說是真正的平行計算。

而在多核的CPU上,則實現了真正意義上的並行。

多執行緒如下圖所示:
這裡寫圖片描述

四、多執行緒安全

多執行緒併發會存在以下三個方面的問題:
(1)最首要的是安全問題:

安全主要在兩個方面:

一是兩個執行緒之間的相互干擾;
當多個執行緒對同一個資料進行操作的時候,如果多個執行緒之間出現了交錯。

比如a++:

首先獲取a的值,然後對其加1,最後將更新的值儲存在a中;

同樣的操作對於b–;

此時有兩個執行緒對a分別進行上述操作,有可能出現的是:

執行緒一獲取a值,執行緒二獲取a值;

執行緒一對a進行+1,執行緒二對a進行-1;

執行緒一更新a值,執行緒二更新a值;

最後a值沒有改變。

二是記憶體一致性,也就是不同的執行緒對同一資料產生了不同的看法。

假設
int a = 0;

執行緒一進行自增操作:

a++;

然後,執行緒二進行輸出操作:

c<<a ;

最後輸出的結果既有可能是0也有可能是1了。

解決方案:

為了解決上述問題,最直接的方法是互斥鎖,就是當一個執行緒對某個資料進行排他性訪問的時候,會得到了其內部鎖之後才能訪問,而在該執行緒沒有釋放此鎖的情況下,其他的執行緒是無法訪問的。

另外一種方式則是阻塞block,當一個執行緒在執行時,另一個執行緒處於休眠的狀態。總而言之,就是將並行轉換為序列來避開競爭的問題。

你以為加了鎖就可以高枕無憂了嗎,naive
因為互斥鎖或者阻塞又會引入新的問題——

(2)執行緒活躍度

首先, 當一個執行緒鎖定了另一個執行緒需要的資源的時候,而第二個執行緒又鎖定了第一個執行緒需要的資源,這個時候就會發生死鎖,雙方都在等待對方先釋放所需資源。

其次,當一個執行緒長時間地佔用某個資源,而其他的執行緒只能被迫長時間地等待。

然後,執行緒1可以使用資源,但是他讓執行緒2先使用,同樣地,執行緒2也在謙讓,致使雙方還是無法進行。

另外,如果對同樣一個資源,多次呼叫非遞迴鎖,會造成多重鎖死;

(3)效能問題-主要是執行緒切換,不作為本文重點。

五、C++11中的多執行緒

C++11將Boost庫中的thread類引進——std::thread,同時提供了std::atomic,std::mutex,std::conditonal_variable,std::future,配上智慧指標,使得c++11中的多執行緒併發變得安全容易。

C++11 所定義的執行緒是和操作系的執行緒是一一對應的,也就是說我們生成的執行緒最終還是直接接受作業系統的排程的。

5.1、建立和結束執行緒

建立一個新執行緒,實質就是定義一個thread類,而對該類的初始化的不同對應不同的幾種方式:
(1)使用一個函式指標作為入口,傳入相關引數

shared_ptr<boost::thread> thread_;

thread_.reset(new thread(funcReturnInt, "%d%s\n", 100, "\%"));

(2)使用C++11中的匿名函式:

thread_.reset(new thread( [](int ia, int ib){
 cout << (ia + ib) << endl;},a,b ));

(3)預設構造,建立空的執行緒:

thread_.reset(new thread());

(4)C++新特性,移動構造,移動構造後t1就析構了:

std::thread t2(std::move(t1))

同時,該類的複製建構函式:thread (const thread&) = delete;意味著該類物件不能被複制構造。
以下是四種構造方式的例項:

#include <iostream>
#include <utility>
#include <thread>//std::thread,std::this_thread
#include <functional>
#include <atomic>

void f1(int n)
{
    for (int i = 0; i < 5; ++i) {
        std::cout << "Thread " 
        << std::this_thread::get_id()
        << " executing\n";
    }
}

int main()
{
    int n = 0;
    std::thread t1; // 空執行緒,實際上也什麼都沒有。
    std::thread t2(f1, n); 
    std::thread t3( [](int ia, int ib){
 cout << (ia + ib) << endl;},a,b );
    std::thread t4(std::move(t2))//移動構造之後,t2已經析構掉了
    t3.join();
    t4.join();
    std::cout << "Final value of n is " << n << '\n';
}

結束一個程序,之前在CAFFE原始碼中講過有兩種方式,一是等待執行緒自己結束,呼叫join()等待目標執行緒結束為止;二是將執行緒分離dispatch(),再主動殺死執行緒,但是C++11不能直接結束執行緒,所以只能被動等待。

5.2、執行緒排程
C++11中沒有排程策略有關的類或者函式

5.3、data_racing

5.3.1 atomic

對於基本資料型別,可以採用原子訪問:
原本需要三步的自增操作,現在必須一口氣完成,中間不能中斷,也就出現不了執行緒干擾的問題。

atomic內容很多,以後再單獨學習。

atomic<int> a(0) ;
thread ta( func_inc, &a);

5.3.2 std::mutex

Mutex 又稱互斥量,提供了獨佔所有權的特性。

mutex有四類:


    std::mutex//最基本的 Mutex 類。
    std::recursive_mutex//遞迴 Mutex 類。
    std::time_mutex//定時 Mutex 類。
    std::recursive_timed_mutex//定時遞迴 Mutex 類。

當我們可能會在某個執行緒內重複呼叫某個鎖的加鎖動作時,我們應該使用遞迴鎖 (recursive lock),除此之外,統統使用最基本的mutex即可。

std::mutex不允許拷貝構造,也不允許 move 拷貝,最初產生的 mutex 物件是處於 unlocked 狀態的。

鎖有兩類:

i、使用std:: lock_guard,lock_guard 是一個範圍鎖,本質是 RAII(Resource Acquire Is Initialization),在構建的時候自動加鎖,在析構的時候自動解鎖,這保證了每一次加鎖都會得到解鎖,但這並不意味著lock_guard 物件能決定 Mutex 物件的生命週期,只負責鎖。

#include <iostream>       
#include <vector>        
#include <thread>        
#include <mutex>          
std::mutex mtx


void print_hello(int label){
    std::lock_guard<std::mutex> lck (mtx);
    std::cout<<"hello thread"<<label<<'\n';
}

int main () {
  std::vector<std::thread> threads;
  for (int i=0; i<10; ++i)
    threads.push_back(std::thread(print_hello,i+1));

  for (auto& th : threads) th.join();
    }

ii、手動加鎖:std::unique_lock 或者直接使用 mutex,mutex中呼叫lock、unlock等:

std::unique_lock 同樣是基於RAII,負責自動加鎖和解鎖,但是他提供了手動上鎖的機會。

std::unique_lock 的建構函式相當多,提供了多種選擇。

其函式方法除了加鎖和解鎖:
lock,
try_lock,如果mutex已經被其他執行緒鎖住某,上鎖失敗,返回false。
try_lock_for,try_lock_until 和 unlock

還有swap():與另一個 std::unique_lock 物件交換它們所管理的 Mutex 物件的所有權;

mutex():返回指向mutex物件指標;
release():返回指向mutex物件指標,釋放所有權
owns_lock():返回是否獲得鎖。

void print_thread_id (int id) {
  std::unique_lock<std::mutex> lck (mtx,std::defer_lock);//此時並沒有鎖住mutex

  lck.lock();//手動加鎖
  std::cout << "thread #" << id << '\n';
  lck.unlock();//手動解鎖
}

try_lock()返回布林值,可以用於判斷。

  if (lck.try_lock())
    std::cout <<"thread #" << id << '\n';
  else                    
    std::cout << 'mutex has been locked';

純手工加鎖

std::mutex mtx;
...
 mtx.lock();
 (*p)++;
 mtx.unlock();
 ...

(3)std::condition_variable
條件變數:條件變數可以讓一個執行緒等待其它執行緒的通知 (wait,wait_for,wait_until),也可以給其它執行緒傳送通知 (notify_one,notify_all),條件變數必須和鎖配合使用,在等待時因為有解鎖和重新加鎖,所以,在等待時必須使用可以手工解鎖和加鎖的鎖,比如 unique_lock,而不能使用 lock_guard。

以下使用CAFFE中的例子:
完成push操作,向另一個執行緒傳送通知:

template<typename T>
class BlockingQueue<T>::sync {
 public:
  mutable boost::mutex mutex_;#互斥鎖
  boost::condition_variable condition_; #條件變數
};


template<typename T>
void BlockingQueue<T>::push(const T& t) {
  boost::mutex::scoped_lock lock(sync_->mutex_);
  queue_.push(t);
  lock.unlock();
  sync_->condition_.notify_one();
}

接收到其他執行緒的通知,立即喚醒該執行緒,否則wait()會自動呼叫 lck.unlock() 釋放鎖,使得其他在等待資源的執行緒可以獲得該鎖:

template<typename T>
T BlockingQueue<T>::pop(const string& log_on_wait) {
  boost::mutex::scoped_lock lock(sync_->mutex_);

  while (queue_.empty()) {
    if (!log_on_wait.empty()) {
      LOG_EVERY_N(INFO, 1000)<< log_on_wait;
    }
    sync_->condition_.wait(lock);
  }

  T t = queue_.front();
  queue_.pop();
  return t;
}

(5)執行緒本地儲存機制:對於共享資源,TSL保證每個執行緒擁有一個資源的副本,然後允許各執行緒訪問各自對應的副本,最後再將副本合併。實現的機制就是建立一個查詢表,根據執行緒的id讀取執行緒對應的那一份資料。
例如CAFFE中:

static boost::thread_specific_ptr<Caffe> thread_instance_;

c++11中:

std::thread_specific_ptr<Caffe> thread_instance_

5.4、活躍度問題

(1)死鎖:
死鎖的條件[2]

資源互斥,某個資源在某一時刻只能被一個執行緒持有 (hold);

吃著碗裡的還看著鍋裡的,持有一個以上的互斥資源的執行緒在等待被其它程序持有的互斥資源;

不可搶佔,只有在某互斥資源的持有執行緒釋放了該資源之後,其它執行緒才能去持有該資源;

環形等待,有兩個或者兩個以上的執行緒各自持有某些互斥資源,並且各自在等待其它執行緒所持有的互斥資源。

(2)第二個問題,使用條件變數喚醒執行緒,在更高層次的佇列中還可以使用對偶模型進行喚醒。

(3)當我們可能會在某個執行緒內重複呼叫某個鎖的加鎖動作時,我們應該使用遞迴鎖 (recursive lock),在 C++11 中,可以根據需要來使用 recursive_mutex,或者 recursive_timed_mutex。

相關推薦

no