c++多線程基礎3(mutex)

整理自：zh.cppreference.com/w/cpp/thread

互斥鎖

互斥算法避免多個線程同時訪問共享資源。這會避免數據競爭，並提供線程間的同步支持。定義於頭文件 <mutex>

互斥鎖有可重入、不可重入之分。C++標准庫中用 mutex 表示不可重入的互斥鎖，用 recursive_mutex 表示可重入的互斥鎖。為這兩個類增加根據時間來阻塞線程的能力，就又有了兩個新的互斥鎖：timed_mutex(不可重入的鎖)、recursive_timed_mutex(可重入的鎖)

C++標准庫的所有mutex都是不可拷貝的，也不可移動

std::mutex：

mutex 類是能用於保護共享數據免受從多個線程同時訪問的同步原語。mutex 提供排他性非遞歸所有權語義。操作：

lock：如果 mutex 未上鎖，則將其上鎖。否則如果已經其它線程 lock，則阻塞當前線程

try_lock：如果 mutex 未上鎖，則將其上鎖。否則返回 false，並不阻塞當前線程

unlock：如果 mutex 被當前線程鎖住，則將其解鎖。否則，是未定義的行為

native_handle：返回底層實現定義的線程句柄

注意：std::mutex 既不可復制亦不可移動

例1：

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

int g_num = 0;//為 g_num_mutex 所保護
std::mutex g_num_mutex;

void slow_increment(int id) {
    for(int i = 0; i < 3; ++i) {
        g_num_mutex.lock();
        ++g_num;
        cout << id << " => " << g_num << endl;
        g_num_mutex.unlock();

        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

int main(void) {
    std::thread t1(slow_increment, 0);
    std::thread t2(slow_increment, 1);
    t1.join();
    t2.join();

// 輸出：
// 0 => 1
// 1 => 2
// 0 => 3
// 1 => 4
// 0 => 5
// 1 => 6

    return 0;
}

例2：

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

std::chrono::milliseconds interval(100);
std::mutex mtex;
int job_shared = 0;//兩個線程都能修改，mtex將保護此變量
int job_exclusive = 0;//只有一個線程能修改

//此線程能修改 jon_shared 和 job_exclusive
void job_1() {
    std::this_thread::sleep_for(interval);//令job_2持鎖

    while(true) {
        //嘗試鎖定 mtex 以修改 job_shared
        if(mtex.try_lock()) {
            cout << "job shared (" << job_shared << ")\n";
            mtex.unlock();
            return;
        } else {
            //不能修改 job_shared
            ++job_exclusive;
            cout << "job exclusive (" << job_exclusive << ")\n";
            std::this_thread::sleep_for(interval);
        }
    }
}

// 此線程只能修改 job_shared
void job_2() {
    mtex.lock();
    std::this_thread::sleep_for(5 * interval);
    ++job_shared;
    mtex.unlock();
}

int main(void) {
    std::thread t1(job_1);
    std::thread t2(job_2);
    t1.join();
    t2.join();

// 輸出：
// job exclusive (1)
// job exclusive (2)
// job exclusive (3)
// job exclusive (4)
// job shared (1)

    return 0;
}

 

std::timed_mutex：

timed_mutex 類是能用於保護數據免受多個線程同時訪問的同步原語。

以類似 mutex 的行為， timed_mutex 提供排他性非遞歸所有權語義。另外，timed_mutex 在 mutex 的基礎上增加了以下兩個操作：

 

try_lock_for()：

函數原型：template< class Rep, class Period >
bool try_lock_for( const std::chrono::duration<Rep,Period>& timeout_duration );

嘗試鎖互斥。阻塞直到經過指定的 timeout_duration 或得到鎖，取決於何者先到來。成功獲得鎖時返回 true ， 否則返回 false 。若 timeout_duration 小於或等於 timeout_duration.zero() ，則函數表現同 try_lock() 。由於調度或資源爭議延遲，此函數可能阻塞長於 timeout_duration 。

標准推薦用 steady_clock 度量時長。若實現用 system_clock 代替，則等待時間亦可能對時鍾調整敏感。

同 try_lock() ，允許此函數虛假地失敗並返回 false ，即使在 timeout_duration 中某點互斥不為任何線程所鎖定。

若此操作返回 true ，則同一互斥上先前的 unlock() 調用同步於（定義於 std::memory_order ）它。若已占有 mutex 的線程調用 try_lock_for ，則行為未定義。

 

try_lock_until(time_point)：

函數原型：template< class Clock, class Duration >
bool try_lock_until( const std::chrono::time_point<Clock,Duration>& timeout_time );

嘗試所互斥。阻塞直至抵達指定的 timeout_time 或得到鎖，取決於何者先到來。成功獲得鎖時返回 true ，否則返回 false 。若已經過 timeout_time ，則此函數表現同 try_lock() 。

使用傾向於 timeout_time 的時鍾，這表示時鍾調節有影響。從而阻塞的最大時長可能小於但不會大於在調用時的 timeout_time - Clock::now() ，依賴於調整的方向。由於調度或資源爭議延遲，函數亦可能阻塞長於抵達 timeout_time 之后。同 try_lock() ，允許此函數虛假地失敗並返回 false ，即使在 timeout_time 前的某點任何線程都不鎖定互斥。

若此操作返回 true ，則同一互斥上先前的 unlock() 調用同步於（定義於 std::memory_order ）它。

若已占有 mutex 的線程調用 try_lock_until ，則行為未定義。

try_lock_for / until可以檢測到死鎖的出現：

if(!try_lock_for(chrono::hours(1)))
{
  throw "出現死鎖!";  
}

例1：

#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
#include <sstream>
 
std::mutex cout_mutex; // 控制到 std::cout 的訪問
std::timed_mutex mutex;
 
void job(int id) 
{
    using Ms = std::chrono::milliseconds;
    std::ostringstream stream;
 
    for (int i = 0; i < 3; ++i) {
        if (mutex.try_lock_for(Ms(100))) {
            stream << "success ";
            std::this_thread::sleep_for(Ms(100));
            mutex.unlock();
        } else {
            stream << "failed ";
        }
        std::this_thread::sleep_for(Ms(100));
    }
 
    std::lock_guard<std::mutex> lock(cout_mutex);
    std::cout << "[" << id << "] " << stream.str() << "\n";
}
 
int main() 
{
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back(job, i);
    }
 
    for (auto& i: threads) {
        i.join();
    }

// 輸出：
// [0] failed failed failed 
// [3] failed failed success 
// [2] failed success failed 
// [1] success failed success

    return 0;
}

例2：

#include <thread>
#include <iostream>
#include <chrono>
#include <mutex>
 
std::timed_mutex test_mutex;
 
void f()
{
    auto now=std::chrono::steady_clock::now();
    test_mutex.try_lock_until(now + std::chrono::seconds(10));
    std::cout << "hello world\n";
}
 
int main()
{
    std::lock_guard<std::timed_mutex> l(test_mutex);
    std::thread t(f);
    t.join();

    return 0;
}

 

遞歸鎖：

在同一個線程中連續 lock 兩次 mutex 會產生死鎖：

一般情況下，如果同一個線程先后兩次調用 lock，在第二次調⽤用時，由於鎖已經被占用，該線程會掛起等待占用鎖的線程釋放鎖，然而鎖正是被自己占用着的，該線程又被掛起而沒有機會釋放鎖，因此 就永遠處於掛起等待狀態了，於是就形成了死鎖(Deadlock)：

#include<iostream> //std::cout
#include<thread>   //std::thread
#include<mutex>    //std::mutex
using namespace std;
mutex g_mutex;

void threadfun1()
{
    cout << "enter threadfun1" << endl;
    // lock_guard<mutex> lock(g_mutex);
    g_mutex.lock();
    cout << "execute threadfun1" << endl;
    g_mutex.unlock();
}

void threadfun2()
{
    cout << "enter threadfun2" << endl;
    // lock_guard<mutex> lock(g_mutex);
    g_mutex.lock();
    threadfun1();
    cout << "execute threadfun2" << endl;
    g_mutex.unlock();
}

int main()
{
    threadfun2(); //死鎖
    return 0;
}
 
// 運行結果:
// enter threadfun2
// enter threadfun1
//就會產生死鎖

 

此時就需要使用遞歸式互斥量 recursive_mutex 來避免這個問題。recursive_mutex 不會產生上述的死鎖問題，只是是增加鎖的計數，但必須確保你 unlock 和 lock 的次數相同，其他線程才可能鎖這個 mutex：

#include<iostream> //std::cout
#include<thread>   //std::thread
#include<mutex>    //std::mutex
using namespace std;

recursive_mutex g_rec_mutex;

void threadfun1()
{
    cout << "enter threadfun1" << endl;
    lock_guard<recursive_mutex> lock(g_rec_mutex);
    cout << "execute threadfun1" << endl;
}

void threadfun2()
{
    cout << "enter threadfun2" << endl;
    lock_guard<recursive_mutex> lock(g_rec_mutex);
    threadfun1();
    cout << "execute threadfun2" << endl;
}

int main()
{
    threadfun2(); //利用遞歸式互斥量來避免這個問題
    return 0;
}
// 運行結果:
// enter threadfun2
// enter threadfun1
// execute threadfun1
// execute threadfun2

recursive_mutex、recursive_timed_mutex 與對應的 mutex、timed_mutex 操作一致。不同點在於，非遞歸鎖在 lock 或 try_lock 一個已經被當前線程 lock 的鎖時會導致死鎖，而遞歸鎖不會

 

共享鎖：

std::shared_timed_mutex(c++14起)

shared_mutex 類是能用於保護數據免受多個線程同時訪問的同步原語。與其他促進排他性訪問的互斥類型相反， shared_mutex 擁有二個層次的訪問：

• 共享 - 多個線程能共享同一互斥的所有權。

• 排他性 - 僅一個線程能占有互斥。

共享互斥通常用於多個讀線程能同時訪問同一資源而不導致數據競爭，但只有一個寫線程能訪問的情形：

#include <iostream>
#include <mutex>  // 對於 std::unique_lock
#include <shared_mutex>
#include <thread>
 
class ThreadSafeCounter {
 public:
  ThreadSafeCounter() = default;
 
  // 多個線程/讀者能同時讀計數器的值。
  unsigned int get() const {
    std::shared_lock<std::shared_timed_mutex> lock(mutex_);//shared_lock 作用類似於 lock_guard
    return value_;
  }
 
  // 只有一個線程/寫者能增加/寫線程的值。
  void increment() {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    value_++;
  }
 
  // 只有一個線程/寫者能重置/寫線程的值。
  void reset() {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    value_ = 0;
  }
 
 private:
  mutable std::shared_timed_mutex mutex_;
  unsigned int value_ = 0;
};
 
int main() {
  ThreadSafeCounter counter;
 
  auto increment_and_print = [&counter]() {
    for (int i = 0; i < 3; i++) {
      counter.increment();
      std::cout << std::this_thread::get_id() << ' ' << counter.get() << '\n';
 
      // 注意：寫入 std::cout 實際上也要由另一互斥同步。省略它以保持示例簡潔。
    }
  };
 
  std::thread thread1(increment_and_print);
  std::thread thread2(increment_and_print);
 
  thread1.join();
  thread2.join();

// 輸出：
// 2 1
// 3 2
// 2 3
// 3 4
// 2 5
// 3 6

  return 0;
}

 

std::shared_mutex(c++17起)

以類似 timed_mutex 的行為， shared_timed_mutex 提供通過 try_lock_for() 、 try_lock_until() 、 try_lock_shared_for() 、 try_lock_shared_until() 方法，試圖帶時限地要求 shared_timed_mutex 所有權的能力。std::shared_mutex 則恰好相反

 

通用互斥管理：

定義於頭文件 <mutex>

std::lock_guard：

類 lock_guard 是互斥封裝器，為在作用域塊期間占有互斥提供便利 RAII 風格機制。

創建 lock_guard 對象時，它試圖接收給定互斥的所有權。控制離開創建 lock_guard 對象的作用域時，銷毀 lock_guard 並釋放互斥。

lock_guard 類不可復制

要鎖定的互斥，類型必須滿足基礎可鎖要求

代碼：

#include <thread>
#include <mutex>
#include <iostream>
 
int g_i = 0;
std::mutex g_i_mutex;  // 保護 g_i
 
void safe_increment()
{
    std::lock_guard<std::mutex> lock(g_i_mutex);
    ++g_i;
 
    std::cout << std::this_thread::get_id() << ": " << g_i << '\n';
 
    // g_i_mutex 在鎖離開作用域時自動釋放
}
 
int main()
{
    std::cout << "main: " << g_i << '\n';
 
    std::thread t1(safe_increment);
    std::thread t2(safe_increment);
 
    t1.join();
    t2.join();
 
    std::cout << "main: " << g_i << '\n';

// 輸出：
// main: 0
// 2: 1
// 3: 2
// main: 2

    return 0;
}

 

std::scoped_lock(c++17起)：

類 scoped_lock 是提供便利 RAII 風格機制的互斥包裝器，它在作用域塊的存在期間占有一或多個互斥。

創建 scoped_lock 對象時，它試圖取得給定互斥的所有權。控制離開創建 scoped_lock 對象的作用域時，析構 scoped_lock 並以逆序釋放互斥。若給出數個互斥，則使用免死鎖算法，如同以 std::lock 。

scoped_lock 類不可復制

要鎖的互斥類型必須滿足可鎖要求

構造函數：

explicit scoped_lock( MutexTypes&... m );

scoped_lock( std::adopt_lock_t, MutexTypes&... m );

 

scoped_lock( const scoped_lock& ) = delete;

取得給定互斥 m 的所有權。

1) 若 sizeof...(MutexTypes) == 0 則不做任何事。否則若 sizeof...(MutexTypes) == 1 ，則等效地調用 m.lock() 。否則，等效地調用 std::lock(m...) 。若 MutexTypes 之一不是遞歸互斥，且當前線程已占有 m... 中對應的參數，則行為未定義。

2) 取得互斥 m... 的所有權而試圖不鎖定任何互斥。若當前線程已占有 m... 中所有互斥，則行為未定義。

3) 復制構造函數被刪除。

若 m 在 scoped_lock 對象之前被銷毀，則行為未定義

要獲得其所有權的互斥

代碼：

#include <mutex>
#include <thread>
#include <iostream>
#include <vector>
#include <functional>
#include <chrono>
#include <string>
 
struct Employee {
    Employee(std::string id) : id(id) {}
    std::string id;
    std::vector<std::string> lunch_partners;
    std::mutex m;
    std::string output() const
    {
        std::string ret = "Employee " + id + " has lunch partners: ";
        for( const auto& partner : lunch_partners )
            ret += partner + " ";
        return ret;
    }
};
 
void send_mail(Employee &, Employee &)
{
    // 模擬耗時的發信操作
    std::this_thread::sleep_for(std::chrono::seconds(1));
}
 
void assign_lunch_partner(Employee &e1, Employee &e2)
{
    static std::mutex io_mutex;
    {
        std::lock_guard<std::mutex> lk(io_mutex);
        std::cout << e1.id << " and " << e2.id << " are waiting for locks" << std::endl;
    }
 
    {
        // 用 std::scoped_lock 取得二個鎖，而無需擔心
        // 其他對 assign_lunch_partner 的調用死鎖我們
        // 而且它亦提供便利的 RAII 風格機制
 
        std::scoped_lock lock(e1.m, e2.m);
 
        // 等價代碼 1 （用 std::lock 和 std::lock_guard ）
        // std::lock(e1.m, e2.m);
        // std::lock_guard<std::mutex> lk1(e1.m, std::adopt_lock);
        // std::lock_guard<std::mutex> lk2(e2.m, std::adopt_lock);
 
        // 等價代碼 2 （若需要 unique_lock ，例如對於條件變量）
        // std::unique_lock<std::mutex> lk1(e1.m, std::defer_lock);
        // std::unique_lock<std::mutex> lk2(e2.m, std::defer_lock);
        // std::lock(lk1, lk2);
        {
            std::lock_guard<std::mutex> lk(io_mutex);
            std::cout << e1.id << " and " << e2.id << " got locks" << std::endl;
        }
        e1.lunch_partners.push_back(e2.id);
        e2.lunch_partners.push_back(e1.id);
    }
 
    send_mail(e1, e2);
    send_mail(e2, e1);
}
 
int main()
{
    Employee alice("alice"), bob("bob"), christina("christina"), dave("dave");
 
    // 在並行線程中指派，因為就午餐指派發郵件消耗很長時間
    std::vector<std::thread> threads;
    threads.emplace_back(assign_lunch_partner, std::ref(alice), std::ref(bob));
    threads.emplace_back(assign_lunch_partner, std::ref(christina), std::ref(bob));
    threads.emplace_back(assign_lunch_partner, std::ref(christina), std::ref(alice));
    threads.emplace_back(assign_lunch_partner, std::ref(dave), std::ref(bob));
 
    for (auto &thread : threads) thread.join();
    std::cout << alice.output() << '\n'  << bob.output() << '\n'
              << christina.output() << '\n' << dave.output() << '\n';
}

blog.csdn.net/zouxinfox/article/details/5848519

 

unique_lock

類 unique_lock 可移動，但不可復制——它滿足可移動構造 (MoveConstructible) 和可移動賦值 (MoveAssignable) 但不滿足可復制構造 (CopyConstructible) 或可復制賦值 (CopyAssignable) 。類 unique_lock 是通用互斥包裝器，允許延遲鎖定、鎖定的有時限嘗試、遞歸鎖定、所有權轉移和與條件變量一同使用。

類 unique_lock 滿足基礎可鎖 (BasicLockable) 要求。若 Mutex 滿足可鎖 (Lockable) 要求，則 unique_lock 亦滿足可鎖 (Lockable) 要求（例如：能用於 std::lock ） ；若 Mutex 滿足定時可鎖 (TimedLockable) 要求，則 unique_lock 亦滿足定時可鎖 (TimedLockable) 要求

要鎖定的互斥類型。類型必須滿足基礎可鎖 (BasicLockable) 要求

代碼：

#include <mutex>
#include <thread>
#include <chrono>
 
struct Box {
    explicit Box(int num) : num_things{num} {}
 
    int num_things;
    std::mutex m;
};
 
void transfer(Box &from, Box &to, int num)
{
    // 仍未實際取鎖
    std::unique_lock<std::mutex> lock1(from.m, std::defer_lock);
    std::unique_lock<std::mutex> lock2(to.m, std::defer_lock);
 
    // 鎖兩個 unique_lock 而不死鎖
    std::lock(lock1, lock2);
 
    from.num_things -= num;
    to.num_things += num;
 
    // 'from.m' 與 'to.m' 互斥解鎖於 'unique_lock' 析構函數
}
 
int main()
{
    Box acc1(100);
    Box acc2(50);
 
    std::thread t1(transfer, std::ref(acc1), std::ref(acc2), 10);
    std::thread t2(transfer, std::ref(acc2), std::ref(acc1), 5);
 
    t1.join();
    t2.join();
}

 

關於 lock_guard 和 unique_lock：

http://lib.csdn.net/article/cplusplus/29099

http://blog.csdn.net/daaikuaichuan/article/details/71022824