C++ 多线程并发编程2 互斥与锁

同步与互斥

同步调用指的是调用的对象在被调用后会阻塞调用方直到被调用方执行完成，而线程天然是异步调用的。但是不同线程的任务片段之间可能会存在严格的先后顺序，例如线程1负责任务A和C，线程2负责任务B，3个任务必须严格按照A->B->C的顺序执行，那么我们就需要在线程1执行任务A的时候阻塞线程2，任务A完成后解除线程2的阻塞并阻塞线程1直到任务B完成。这样维护和协调线程任务片段之间先后顺序的操作，就叫做线程间的同步。

而互斥就是保证资源在同一时刻只能被一个进程所使用，确保数据的一致性。

互斥量 mutex

前面我们提到，同一个进程中不同的线程会共享这个进程的地址空间等资源，进程之间也可能存在共享的资源（例如共享内存），这就可能出现不同线程或进程同时使用同一共享资源的访问冲突，我们称之为竞态条件(data race)。为了解决这一冲突，我们引入互斥量，为资源加锁解锁以确保互斥。

#include <mutex>

C++ 11中有4种mutex类

• std::mutex，最基本的 Mutex 类。
• std::recursive_mutex，递归 Mutex 类。
• std::time_mutex，定时 Mutex 类。
• std::recursive_timed_mutex，定时递归 Mutex 类。

除了mutex类外，头文件<mutex>中还包含

• 两种lock类
  • std::lock_guard
  • std::unique_lock
• 与互斥锁相关的函数
  • std::try_lock，尝试同时对多个互斥量上锁。
  • std::lock，可以同时对多个互斥量上锁。
  • std::call_once，如果多个线程需要同时调用某个函数，call_once 可以保证多个线程对该函数只调用一次。

std::mutex

是C++11中最基本的互斥量,包含下列成员函数

• 构造函数
  std::mutex不允许拷贝构造和移动构造，只能无参地声明一个初始状态为unlock的mutex对象
• lock()
  将该互斥锁锁住。线程调用该函数会发生下面 3 种情况：
  1. 如果该互斥量当前没有被锁住，则调用线程将该互斥量锁住，直到调用 unlock之前，该线程一直拥有该锁。
  2. 如果当前互斥量被其他线程锁住，则当前的调用线程被阻塞住。
  3. 如果当前互斥量被当前调用线程锁住，则会产生死锁(deadlock)
• unlock()
  将互斥锁解锁，并释放对互斥锁的所有权
• try_lock()
  尝试锁住互斥锁，与lock()的区别在于，如果互斥量被其他线程占有，则当前线程也不会被阻塞，而是返回false。

示例

#include <iostream>      
#include <thread>        
#include <mutex>         

volatile int counter(0); 
std::mutex mtx;           

void attempt_10k_increases() {
    for (int i=0; i<10000; ++i) {
        if (mtx.try_lock()) {  
            counter++;
            mtx.unlock();
        }        
    }
}

int main (int argc, const char* argv[]) {
    std::thread threads[10];
    for (int i=0; i<10; ++i) {
        threads[i] = std::thread(attempt_10k_increases);
    }
        

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

    std::cout << counter << " successful increases of the counter.\n";

    return 0;
}

运行上面的代码，会发现输出的值不是100000，这是因为函数attempt_10k_increases只有在线程拥有互斥锁的所有权时才会将counter++，如果不拥有所有权也不会阻塞而是继续运行for循环。如果我们将attempt_10k_increases改成

void attempt_10k_increases() {
    for (int i=0; i<10000; ++i) {
        mtx.lock();
        counter++;
        mtx.unlock();
    }
}

那么最后的counter就会是100000。

std::recursive_mutex

std::recursive_mutex与std::mutex的不同之处在于

• 允许同一个线程对互斥量多次上锁（即递归上锁），来获得对互斥量对象的多层所有权
• 释放互斥量时需要调用与该锁层次深度相同次数的 unlock()，即调用unlock的次数需要与调用lock（或try_lock成功）的次数相等

std::time_mutex

std::time_mutex比std::mutex多了下面两个函数

• try_lock_for
  接受一个时间范围，表示在这一段时间范围之内线程如果没有获得锁则被阻塞住，如果在此期间其他线程释放了锁，则该线程可以获得对互斥量的锁，如果超时（即在指定时间内还是没有获得锁），则返回false。
• try_lock_until
  作用与try_lock_for相同，区别在于接受一个时间点而非时间范围作为参数。

std::recursive_timed_mutex

能够指定时间范围或时间节点的递归mutex

互斥锁的RAII管理

std::lock_guard

是C++中的互斥锁管理类，用RAII的方式自动管理互斥锁的加锁与解锁，避免忘记释放锁导致死锁或资源的泄露。

• 构造时对给定的std::mutex加锁，析构时自动解锁
• 不能手动加锁或解锁
• 适用于作用域内始终保持锁定的场景
• 禁用拷贝构造和移动构造

std::unique_lock

与std::lock_guard的区别在于更加灵活，提供了更多的功能。

• 构造函数

  • explicit unique_lock(Mutex& m);

    std::mutex mtx;
    // 构造时立即加锁，析构时自动解锁
    std::unique_lock<std::mutex> lk(mtx);

  • unique_lock(Mutex& m, std::defer_lock_t);

    std::mutex mtx;
    // 构造时不加锁，可延迟到后面手动 lock()
    std::unique_lock<std::mutex> lk(mtx, std::defer_lock);
    // … 
    lk.lock();

  • unique_lock(Mutex& m, std::try_to_lock_t);

    std::mutex mtx;
    // 构造时尝试加锁，成功则 owns_lock()==true
    std::unique_lock<std::mutex> lk(mtx, std::try_to_lock);
    if (lk.owns_lock()) { /* 已加锁 */ }

  • unique_lock(Mutex& m, std::adopt_lock_t);

    std::mutex mtx;
    mtx.lock();  
    // 构造时假设锁已被外部加锁，仅管理解锁
    std::unique_lock<std::mutex> lk(mtx, std::adopt_lock);
    // 析构时才会调用 mtx.unlock()

  • unique_lock(Mutex& m, const std::chrono::duration<Rep,Period>& rel_time);

    std::timed_mutex tmtx;
    // 最多等待 100ms 进行加锁
    std::unique_lock<std::timed_mutex> lk(tmtx, std::chrono::milliseconds(100));
    if (lk.owns_lock()) { /* 加锁成功 */ }

• 加锁与解锁

  • .lock()

    lk.lock();  // 阻塞直到加锁成功，否则抛 system_error

  • .try_lock()

    if (lk.try_lock()) {
        // 立即返回，加锁成功或失败均不阻塞
    }

  • .try_lock_for(...) / .try_lock_until(...)

    // 在给定时长或到达某时刻前尝试加锁
    if (lk.try_lock_for(std::chrono::milliseconds(50))) { /* 成功 */ }

  • .unlock()

    lk.unlock();  // 释放锁，之后 owns_lock()==false

• 状态查询

  • .owns_lock()

    if (lk.owns_lock()) { /* 当前持有锁 */ }

  • operator bool()

    if (lk) { /* 等同于 owns_lock() */ }

  • .mutex()

    std::mutex* p = lk.mutex();  // 获取绑定的互斥量指针

• 所有权转移与释放

  • 移动构造 / 移动赋值

    std::unique_lock<std::mutex> a(mtx);
    std::unique_lock<std::mutex> b(std::move(a));
    // b 接管了 a 的锁，a.owns_lock()==false, a.mutex()==nullptr

  • .release()

    std::mutex* raw = lk.release();
    // 放弃对互斥量的管理但不解锁，后续需要手动 unlock(raw)

• 与条件变量condition_variable配合

  std::mutex mtx;
  std::condition_variable cv;
  bool ready = false;
  
  // 生产者
  void producer() {
      std::unique_lock<std::mutex> lk(mtx);
      ready = true;
      lk.unlock();         // 手动解锁后再通知
      cv.notify_one();
  }
  
  // 消费者
  void consumer() {
      std::unique_lock<std::mutex> lk(mtx);
      cv.wait(lk, []{ return ready; });
      // 等待过程中会自动 unlock/lock
      // 唤醒后持有锁，可安全访问共享数据
  }

死锁

死锁指的是多个线程争夺共享资源导致每个线程都不能取得自己所需的全部资源，从而导致程序无法向下执行。
死锁的产生有4个必要条件

1. 互斥 即请求的资源同一时刻只能被一个进程或线程使用
2. 请求并保持 进程在持有了至少一个资源的同时，又提出了对其它资源的请求，并在请求不到时一直处于等待状态，但不会释放已持有的资源。
3. 不剥夺 已经分配给某进程的资源，只有当该进程自行释放时才会收回；系统不能强制地剥夺该进程的资源。
4. 循环等待 存在一个进程集合 {P₁, P₂, …, Pₙ}，其中 P₁ 等待 P₂ 占有的资源，P₂ 等待 P₃ 占有的资源，…，Pₙ 等待 P₁ 占有的资源，形成一个闭环。

只有当这4个条件全部成立时，才可能发生死锁。

参考

• [知乎专栏]C++多线程并发基础入门教程
• [cnblogs.com]C++11 并发指南系列