RAII手法封装的互斥器mutex和条件变量condition类

前言

近来在学习陈硕老师的muduo库，阅读了里面RAII手法封装的线程安全互斥锁的源码，期间遇到很多问题，包括有些宏对新手非常不友好等，解决这些问题花了很多时间，结合源码和自己的思考以及查阅的资料，本文记录下相关的难点，如有错误，欢迎指正交流。

概要

首先谈谈什么是RAII，用自己的话说RAII的做法是使用一个对象，在其构造时获取资源，在对象生命期控制对资源的访问使之始终保持有效，最后在对象析构的时候释放资源。RAII在C++中应用很广泛，用于管理资源、避免内存泄露。

原则

muduo库遵循以下几个原则：

• 用RAII手法封装mutex的创建、销毁、加锁、解锁这四个操作。
• 只用非递归的mutex（即不可重入的mutex）
• 不手工调用lock()和unlock()函数，一切交给栈上的Guard对象的构造和析构函数负责。
• 每次构造Guard对象的时候，思考一路上（调用栈上）已经持有的锁，防止因加锁顺序不同而导致死锁（deadlock）。由于Guard对象是栈上对象，看函数调用栈就能分析用锁的情况（后面会讲如何查看）。
• 不使用跨进程的mutex，进程间通信只用TCP sockets
• 必要时可以考虑用PTHREAD_MUTEX_ERRORCHECK来排错

宏定义部分

首先说明GUARDED_BY（mutex_）这种用于clang编译器线程安全检查的部分不在本文讨论范围，如果需要详细了解，见此链接Thread Safety Analysis。

Let's go

#ifdef CHECK_PTHREAD_RETURN_VALUE

#ifdef NDEBUG
__BEGIN_DECLS
extern void __assert_perror_fail (int errnum,
                                  const char *file,
                                  unsigned int line,
                                  const char *function)
    noexcept __attribute__ ((__noreturn__));
__END_DECLS
#endif

__attribute__ ((__noreturn__))告诉编译器__assert_perror_fail函数没有返回值也没有问题，__noreturn__的具体解释可以看_ attribute__((noreturn))的用法

#define MCHECK(ret) ({ __typeof__ (ret) errnum = (ret);         
                       if (__builtin_expect(errnum != 0, 0))    
                         __assert_perror_fail (errnum, __FILE__, __LINE__, __func__);})

#else  // CHECK_PTHREAD_RETURN_VALUE

#define MCHECK(ret) ({ __typeof__ (ret) errnum = (ret);         
                       assert(errnum == 0); (void) errnum;})

#endif // CHECK_PTHREAD_RETURN_VALUE

这里的__typeof__相当于typeof()，__typeof__(ret) errnum就是把errnum定义为ret这个变量的类型，而__builtin_expect是一种优化手段，__builtin_expect(A,b)意思就是A表达式大概率等于b，经常被用来做以下宏定义：

#define likely(x) __builtin_expect(!!(x), 1) //x很可能为真       
#define unlikely(x) __builtin_expect(!!(x), 0) //x很可能为假

因为处理器都是流水线的，有些里面有多个逻辑运算单元，系统可以提前取多条指令进行并行处理，但遇到跳转时，则需要重新取指令，__builtin_expect的目的是增加条件分支预测的准确性，cpu 会提前装载后面的指令，遇到条件转移指令时会提前预测并装载某个分支的指令。

__assert_perror_fail (errnum, __FILE__, __LINE__, __func__)
最后会按照FILE ：LINE　： func　: errnum打印出错误，和assert一样，都在编译时候执行，但是好处在于它能按照格式打印出错误在哪里，便于快速排查出错误。

想深入理解流水线优化的看这篇博客__builtin_expect

（ps：真是个神奇的东西。。。）

代码段最后一行也有一个宏，

#define MutexLockGuard(x) static_assert(false, "missing mutex guard var name");

这个宏的作用使防止程序出现如下错误：

void doit()
{
    MutexLockGuard(mutex);	//遗漏变量名，产生一个临时对象又马上销毁了
    						//结果没有锁住临界区
    //正确写法: MutexLockGuard lock(mutex);
}

互斥锁（Mutex）

代码正文部分相对来说没什么难度，在代码里有相关注释。

这里解释下为什么要在MutexLock中还要引入UnassignGuard类，因为MutexLock类在封装mutex_的同时，还额外引入了一个状态成员MutexLock::holder_，这个成员用于记录锁的持有者的线程tid。

当其它线程调用pthread_cond_signal或pthread_cond_signal,它会重写获取锁。

这个时候，pthread_cond_wait势必会改变pthread_mutex_t和MutexLock:holder的一致性。所以需要在调用pthread_cond_wait的前后添加一些代码去相应的修改MutexLock::holder，也就是分别调用MutexLock::unassignHolder和MutexLock::assignHolder。MutexLock::UnassignGuard类的作用，就是利用RAII简化对MutexLock::unassignHolder和MutexLock::assignHolder的调用。

class MutexLock : noncopyable   //noncopyable类利用了c++11中新特性
    							//将拷贝构造函数和复制构造函数delete
    							//确保MutexLock类就不可重入了
{
 public:
  MutexLock()
    : holder_(0)
  {
    MCHECK(pthread_mutex_init(&mutex_, NULL));
        //初始化互斥锁，并检查（MCHECK）是否初始化成功。
  }

  ~MutexLock()
  {
    assert(holder_ == 0);	//持有者不为0则出错，不能销毁锁。
    MCHECK(pthread_mutex_destroy(&mutex_));
  }

  // must be called when locked, i.e. for assertion
  bool isLockedByThisThread() const
  {
    return holder_ == CurrentThread::tid(); //该锁是否被当前线程持有
  }

  void assertLocked() const 
  {
    assert(isLockedByThisThread());
  }

  // internal usage

  void lock() ACQUIRE()
  {
    MCHECK(pthread_mutex_lock(&mutex_));	//上锁，并检查
    assignHolder();	//并注册当前线程的tid
  }

  void unlock() RELEASE()
  {
    unassignHolder();	//清除已经注册的此线程的tid
    MCHECK(pthread_mutex_unlock(&mutex_)); //释放锁
  }

  pthread_mutex_t* getPthreadMutex() /* non-const */
  {
    return &mutex_;
  }

 private:
  friend class Condition;

    //由于存在holder_，用于解除注册，这里结合下面的condition类会更加好理解
  class UnassignGuard : noncopyable 
  {
   public:
    explicit UnassignGuard(MutexLock& owner)
      : owner_(owner)
    {
      owner_.unassignHolder();
    }

    ~UnassignGuard()
    {
      owner_.assignHolder();
    }

   private:
    MutexLock& owner_;
  };

  void unassignHolder()
  {
    holder_ = 0;
  }

  void assignHolder()
  {
    holder_ = CurrentThread::tid();
  }

  pthread_mutex_t mutex_;
  pid_t holder_;
};

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

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

 private:
  MutexLock& mutex_;
};

}  // namespace muduo

#define MutexLockGuard(x) error "Missing guard object name"

条件变量（Condition variable）

1、注意区分signal和broadcast：

broadcast通常用于表明状态变化，signal通常用于表示资源可用

2、pthread_cond_wait：阻塞等待条件变量成立->解锁->加锁（见下图）

• 传入前锁mutex目的：为了保证线程从条件判断到进入pthread_cond_wait前，条件不被改变。
• 传入后解锁的目的：为了条件能够被改变
• 返回前再次锁mutex：返回前再次锁mutex是为了保证线程从pthread_cond_wait返回后 到 再次条件判断前不被改变。

知道原理后Condition类的源码实现也不难。

class Condition : noncopyable
{
 public:
  explicit Condition(MutexLock& mutex)
    : mutex_(mutex)
  {
    MCHECK(pthread_cond_init(&pcond_, NULL));
  }

  ~Condition()
  {
    MCHECK(pthread_cond_destroy(&pcond_));
  }

  void wait()
  {
    MutexLock::UnassignGuard ug(mutex_);  //先将holder_清零，防止出现死锁
    MCHECK(pthread_cond_wait(&pcond_, mutex_.getPthreadMutex()));
     //ug析构的时候，会将holder_置为该线程的tid
  }

  // returns true if time out, false otherwise.
  bool waitForSeconds(double seconds);

  void notify()
  {
    MCHECK(pthread_cond_signal(&pcond_));
  }

  void notifyAll()
  {
    MCHECK(pthread_cond_broadcast(&pcond_));
  }

 private:
  MutexLock& mutex_;
  pthread_cond_t pcond_;
};

}  // namespace muduo

倒计时（CountDownLatch）

一种常用且易用的同步手段，它有两个用途：

• 主线程发起多个子线程，等这些子线程各自都完成一定的任务之后，主线程才继续执行。通常用于主线程等待多个子线程完成初始化。
• 主线程发起多个子线程，子线程都等待主线程，主线程完成其他一些任务之后通知所有子线程开始执行。通常用于多个子线程等待主线程发起“起跑”命令。

当然我们可以直接用条件变量来实现以上两种同步，但如果用CountDownLatch的话，逻辑更加清晰。

class CountDownLatch : boost::noncopyable
{
public:
	explicit CountDownLatch(int count);  //倒数几次
	void wait();					//等待计数值变为0
	void countDown(); 				//计数减一
private:
	mutable MutexLock mutex_;
	Condition condition_;
	int count_;
};

inline 
CountDownLatch::CountDownLatch(int count)
    : mutex_(),
	condition_(mutex_),
	count_(count)
    {}

void CountDownLatch::wait()
{
	MutexLockGuard lock(mutex_);
	while(count_ > 0)
		condition_.wait();
}

void CountDownLatch::countDown()
{
	MutexLockGuard lock(mutex_);
	--count_;
	if(count_ == 0)
		condition_.notifyAll();
}

注意到countDown()使用的是notifyAll()，而前面enqueue()使用的是notify()，原因见condition部分的第一点

死锁调试

死锁比较容易debug，把各个线程的调用栈打出来（gdb中使用thread apply all bt命令），只要每个函数不是特别长，很容易看出来是怎么死的。或者可以用PHTREAD_MUTEX_ERRORCHECK一下子就能找到错误。

具体调试操作可以看陈硕《Linux多线程服务端编程：使用muduo C++网络库》的2.1.2节。

小结

对于线程同步，掌握mutex和condition是最基本的，按陈硕老师的原话MutexLock和MutexLockGuard这两个class应该能在纸上默写出来。

同时看了这么多，还是留下一个小问题，CurrentThread::tid()函数，将在之后再解决。

以上