【多线程】c++11多线程编程(三)——竞争条件与互斥锁

竞争条件

并发代码中最常见的错误之一就是竞争条件(race condition)。而其中最常见的就是数据竞争(data race)，从整体上来看，所有线程之间共享数据的问题，都是修改数据导致的，如果所有的共享数据都是只读的，就不会发生问题。但是这是不可能的，大部分共享数据都是要被修改的。

而c++中常见的cout就是一个共享资源，如果在多个线程同时执行cout，你会发发现很奇怪的问题：

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

// 普通函数 无参
void function_1() {
    for(int i=0; i>-100; i--)
        cout << "From t1: " << i << endl;
}

int main()
{
    std::thread t1(function_1);

    for(int i=0; i<100; i++)
        cout << "From main: " << i << endl;

    t1.join();
    return 0;
}

你有很大的几率发现打印会出现类似于From t1: From main: 64这样奇怪的打印结果。cout是基于流的，会先将你要打印的内容放入缓冲区，可能刚刚一个线程刚刚放入From t1:，另一个线程就执行了，导致输出变乱。而c语言中的printf不会发生这个问题。

使用互斥元保护共享数据

解决办法就是要对cout这个共享资源进行保护。在c++中，可以使用互斥锁std::mutex进行资源保护，头文件是#include <mutex>，共有两种操作：锁定(lock)与解锁(unlock)。将cout重新封装成一个线程安全的函数：

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

std::mutex mu;
// 使用锁保护
void shared_print(string msg, int id) {
    mu.lock(); // 上锁
    cout << msg << id << endl;
    mu.unlock(); // 解锁
}

void function_1() {
    for(int i=0; i>-100; i--)
        shared_print(string("From t1: "), i);
}

int main()
{
    std::thread t1(function_1);

    for(int i=0; i<100; i++)
        shared_print(string("From main: "), i);

    t1.join();
    return 0;
}

修改完之后，运行可以发现打印没有问题了。但是还有一个隐藏着的问题，如果mu.lock()和mu.unlock()之间的语句发生了异常，会发生什么？unlock()语句没有机会执行！导致导致mu一直处于锁着的状态，其他使用shared_print()函数的线程就会阻塞。

解决这个问题也很简单，使用c++中常见的RAII技术，即获取资源即初始化(Resource Acquisition Is Initialization)技术，这是c++中管理资源的常用方式。简单的说就是在类的构造函数中创建资源，在析构函数中释放资源，因为就算发生了异常，c++也能保证类的析构函数能够执行。我们不需要自己写个类包装mutex，c++库已经提供了std::lock_guard类模板，使用方法如下：

void shared_print(string msg, int id) {
    //构造的时候帮忙上锁，析构的时候释放锁
    std::lock_guard<std::mutex> guard(mu);
    //mu.lock(); // 上锁
    cout << msg << id << endl;
    //mu.unlock(); // 解锁
}

可以实现自己的std::lock_guard，类似这样：

class MutexLockGuard
{
 public:
  explicit MutexLockGuard(std::mutex& mutex)
    : mutex_(mutex)
  {
    mutex_.lock();
  }

  ~MutexLockGuard()
  {
    mutex_.unlock();
  }

 private:
  std::mutex& mutex_;
};

template<typename Mutex>

class lock_guard
{

private:

    Mutex& m;

    explicit lock_guard(lock_guard&);

    lock_guard& operator=(lock_guard&);

public:

    explicit lock_guard(Mutex& m_) : m(m_) {

        m.lock();

    }

    lock_guard(Mutex& m_,adopt_lock_t) : m(m_)

    {}

    ~lock_guard() {

        m.unlock();
    }

}

 

为保护共享数据精心组织代码

上面的std::mutex互斥元是个全局变量，他是为shared_print()准备的，这个时候，我们最好将他们绑定在一起，比如说，可以封装成一个类。由于cout是个全局共享的变量，没法完全封装，就算你封装了，外面还是能够使用cout，并且不用通过锁。

(cout 不是多线程安全的，需要自己实现同步，printf是多线程安全)

下面使用文件流举例：

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

std::mutex mu;
class LogFile {
    std::mutex m_mutex;
    ofstream f;
public:
    LogFile() {
        f.open("log.txt");
    }
    ~LogFile() {
        f.close();
    }
    void shared_print(string msg, int id) {
        std::lock_guard<std::mutex> guard(mu);
        f << msg << id << endl;
    }
};

void function_1(LogFile& log) {
    for(int i=0; i>-100; i--)
        log.shared_print(string("From t1: "), i);
}

int main()
{
    LogFile log;
    std::thread t1(function_1, std::ref(log));

    for(int i=0; i<100; i++)
        log.shared_print(string("From main: "), i);

    t1.join();
    return 0;
}

上面的LogFile类封装了一个mutex和一个ofstream对象，然后shared_print函数在mutex的保护下，是线程安全的。使用的时候，先定义一个LogFile的实例log，主线程中直接使用，子线程中通过引用传递过去(也可以使用单例模式来实现),这样就能保证资源被互斥锁保护着，外面没办法使用但是使用资源。

但是这个时候还是得小心了！用互斥元保护数据并不只是像上面那样保护每个函数，就能够完全的保证线程安全，如果将资源的指针或者引用不小心传递出来了，所有的保护都白费了！要记住一下两点：

1. 不要提供函数让用户获取资源。

   std::mutex mu;
   class LogFile {
       std::mutex m_mutex;
       ofstream f;
   public:
       LogFile() {
           f.open("log.txt");
       }
       ~LogFile() {
           f.close();
       }
       void shared_print(string msg, int id) {
           std::lock_guard<std::mutex> guard(mu);
           f << msg << id << endl;
       }
       // Never return f to the outside world
       ofstream& getStream() {
           return f;  //never do this !!!
       }
   };

2. 不要资源传递给用户的函数。

   class LogFile {
       std::mutex m_mutex;
       ofstream f;
   public:
       LogFile() {
           f.open("log.txt");
       }
       ~LogFile() {
           f.close();
       }
       void shared_print(string msg, int id) {
           std::lock_guard<std::mutex> guard(mu);
           f << msg << id << endl;
       }
       // Never return f to the outside world
       ofstream& getStream() {
           return f;  //never do this !!!
       }
       // Never pass f as an argument to user provided function
       void process(void fun(ostream&)) {
           fun(f);
       }
   };

以上两种做法都会将资源暴露给用户，造成不必要的安全隐患。

接口设计中也存在竞争条件

STL中的stack类是线程不安全的，于是你模仿着想写一个属于自己的线程安全的类Stack。于是，你在push和pop等操作得时候，加了互斥锁保护数据。但是在多线程环境下使用使用你的Stack类的时候，却仍然有可能是线程不安全的，why？

假设你的Stack类的接口如下：

class Stack
{
public:
    Stack() {}
    void pop(); //弹出栈顶元素
    int& top(); //获取栈顶元素
    void push(int x);//将元素放入栈
private:
    vector<int> data; 
    std::mutex _mu; //保护内部数据
};

类中的每一个函数都是线程安全的，但是组合起来却不是。加入栈中有9,3,8,6共4个元素，你想使用两个线程分别取出栈中的元素进行处理，如下所示：

Thread A               Thread B
int v = st.top(); // 6
                      int v = st.top(); // 6
st.pop(); //弹出6
                      st.pop(); //弹出8
                      process(v);//处理6
process(v); //处理6

top()是取出栈顶元素，不会删掉栈里边的元素

pop()是删除栈顶元素

可以发现在这种执行顺序下， 栈顶元素被处理了两遍，而且多弹出了一个元素8，导致`8没有被处理！这就是由于接口设计不当引起的竞争。解决办法就是将这两个接口合并为一个接口！就可以得到线程安全的栈。

class Stack
{
public:
    Stack() {}
    int& pop(); //弹出栈顶元素并返回
    void push(int x);//将元素放入栈
private:
    vector<int> data; 
    std::mutex _mu; //保护内部数据
};

//下面这样使用就不会发生问题
int v = st.pop(); // 6
process(v);

但是注意：这样修改之后是线程安全的，但是并不是异常安全的，这也是为什么STL中栈的出栈操作分解成了两个步骤的原因。(为什么不是异常安全的还没想明白。。)

所以，为了保护共享数据，还得好好设计接口才行。

参考

1. C++并发编程实战
2. C++ Threading #3: Data Race and Mutex

锁的概念与使用

C++11 提供了几种锁，最常用的就是 std::mutex ，把它放到需要保护的代码段之前，调用它的 lock 方法实现加锁， unlock 实现解锁，加锁和解锁必须成对出现，否则会出现死锁的问题。

为了防止使用者忘记调用 unlock ，c++11 还提供了 scope_lock 和 unique_lock 两个对象来实现自动解锁，无须显式调用 unlock 了。

在 code3 目录下新建一个 code5.cpp 文件：

#include <iostream>
#include <functional>
#include <thread>
#include <memory>
#include <mutex>

class counter{
public:
    void increase(){
        std::this_thread::sleep_for(std::chrono::microseconds(20));
        std::cout<<"thread id: "<<std::this_thread::get_id()<<std::endl;
        mtx_.lock();
        counter_++;
        mtx_.unlock();
    }

    void increase1(){
        std::this_thread::sleep_for(std::chrono::microseconds(20));
        std::cout<<"thread id: "<<std::this_thread::get_id()<<std::endl;
        std::unique_lock<std::mutex> lock(mtx_);
        counter_++;
    }

    void print_counter() {
        std::unique_lock<std::mutex> lock(mtx_);
        std::cout<<"thread id: "<<std::this_thread::get_id()<<" counter: "<<counter_<<std::endl;
    }

private:
    int counter_ = 0;
    std::mutex mtx_;
};

int main(){
    counter ct;

    std::thread thd1([&ct]{
        ct.increase();
    });

    std::thread thd2([&ct]{
        ct.increase1();
    });

    std::thread thd3([&ct]{
        ct.print_counter();
    });

    thd1.join();
    thd2.join();
    thd3.join();
}

编译和运行代码：在 build 目录下执行

g++ ../code5.cpp -o code5 -std=c++11 -lpthread && ./code5

输出结果不唯一，以下是我的输出：

thread id: 140125434894080 counter: 0
thread id: 140125443286784
thread id: 140125451679488

推荐使用 std::unique_lock<std::mutex> lock(mtx_) 这种写法，可以保证不会忘记解锁。

std::mutex 锁上的并不是变量，所以上述代码并没有锁和变量绑定的操作。 std::mutex 其实是一种执行权的象征，当任何一线程的代码执行到 mtx_.lock(); 的时候，就开始尝试申请执行权，如果成功申请到，则可以继续执行，否则阻塞等待。有执行权的线程在 mtx_.unlock(); 之前，其他任何进行执行到 mtx_.lock(); 都会阻塞等待。取得执行权后执行的代码是任意的，可以对互斥变量进行更改，也可以什么都不做。

原子变量的概念与使用

C++11 还提供了原子变量来保证线程安全，对于一个原子变量来说对它的多线程操作是安全的。

对于仅仅是整形或者 bool 类型的多线程访问，我们可以使用原子变量来保证线程安全，比加锁的写法更加简单。

在 code3 目录下新建一个 code6.cpp 文件：

#include <iostream>
#include <functional>
#include <thread>
#include <memory>
#include <atomic>

class counter {
public:
    counter() : counter_(0) {};
    void increase() {
        counter_++;
    }

    void print_counter() {
        std::cout << "thread id: " << std::this_thread::get_id() << " counter: " << counter_ << std::endl;
    }

private:
    std::atomic<int> counter_;
};

int main() {
    counter ct;

    std::thread thd1([&ct] {
        ct.increase();
        });

    std::thread thd2([&ct] {
        ct.increase();
        });

    std::thread thd3([&ct] {
        ct.print_counter();
        });

    thd1.join();
    thd2.join();
    thd3.join();
}

编译和运行代码：在 build 目录下执行

g++ ../code6.cpp -o code6 -std=c++11 -lpthread && ./code6

输出结果：

thread id: 139703679284992 counter: 2

1. std::thread 创建和运行线程
2. std::join 阻塞等待线程退出
3. std::detch 让线程独立运行
4. 线程组可以在一定程度上提高程序运行效率
5. 原子变量与锁