Muduo库之同步原语

标签：__ Muduo void mu 原语同步线程 type ptr

Atomic

在 Atomic.h 文件中定义了原子操作类类型 AtomicIntegerT<T>，它使用了 GCC 内置的原子操作来实现。

原子操作在多线程开发中经常用到，比如计数器、序列产生器等。这些情况下数据有并发的危险，但是用锁去保护又显得有些浪费，会造成一定的性能瓶颈，而原子操作则更为节省资源，所以原子操作最为合适。

比如，常见的 i++ 操作便不是原子操作，它通常分为以下三步：

从内存将变量的数值取到寄存器中；
对寄存器中的数值加1；
将寄存器中的数值存入到内存中；

由于时序的问题，所以如果多个线程同时操作同一个全局变量，就会出现问题。

这里所说的原子操作，基本包含如下语义：

操作本身是不可分割的；
一个线程对某个数据的操作何时对另外一个线程可见；
执行的顺序是否可以被重排；

GCC 自 4.1.2 版本之后，对 X86 或 X86_64 支持内置原子操作。也就是说，无需引入第三方库的锁保护，便可对 1、2、4、8 字节的数值或指针类型进行原子加/减/与/或/异或等操作。__sync_ 系列函数如下：

// 将 value 加到 ptr 上，结果更新到 ptr，并返回操作之前的 ptr 的值
type __sync_fetch_and_add(type* ptr, type value, ...);
// 从 ptr 减去 val，结果更新到 ptr，并返回操作之前的 ptr 的值
type __sync_fetch_and_sub(type* ptr, type value, ...);
// 将 ptr 与 value 相或，结果更新到 ptr，并返回操作之前的 ptr 的值
type __sync_fetch_and_or(type* ptr, type value, ...);
// 将 ptr 与 value 相与，结果更新到 ptr，并返回操作之前的 ptr 的值
type __sync_fetch_and_and(type* ptr, type value, ...);
// 将 ptr 与 value 异或，结果更新到 ptr，并返回操作之前的 ptr 的值
type __sync_fetch_and_xor(type* ptr, type value, ...);
// 将 ptr 取反后，与 value 相与，结果更新到 ptr，并返回操作之前的 ptr 的值
type __sync_fetch_and_nand(type* ptr, type value, ...);
// 将 value 加到 ptr 上，结果更新到 ptr，并返回操作之后新的 ptr 的值
type __sync_add_and_fetch(type* ptr, type value, ...);
// 从 ptr 减去 val，结果更新到 ptr，并返回操作之后新的 ptr 的值
type __sync_sub_and_fetch(type* ptr, type value, ...);
//  将 ptr 与 value 相或，结果更新到 ptr，并返回操作之后新的 ptr 的值
type __sync_or_and_fetch(type* ptr, type value, ...);
// 将 ptr 与 value 相与，结果更新到 ptr，并返回操作之后新的 ptr 的值
type __sync_and_and_fetch(type* ptr, type value, ...);
// 将 ptr 与 value 异或，结果更新到 ptr，并返回操作之后新的 ptr 的值
type __sync_xor_and_fetch(type* ptr, type value, ...);
// 将 ptr 取反后，与 value 相与，结果更新到 ptr，并返回操作之后新的 ptr 的值
type __sync_nand_and_fetch(type* ptr, type value, ...);
// 比较 ptr 与 oldval 的值，如果两者相等，则将 newval 更新到 ptr 并返回 true
bool __sync_bool_compare_and_swap(type* ptr, type oldval, type newval ...);
// 比较 ptr 与 oldval 的值，如果两者相等，则将 newval 更新到 ptr 并返回操作之前 ptr 的值
type __sync_val_compare_and_swap(type* ptr, type oldval, type newval, ...);
// 屏障函数
__sync_synchronize(...);
// 将 value 写入 ptr，对 ptr 加锁，并返回操作之前 ptr 的值，即 try spinlock 语义
type __sync_lock_test_and_set(type *ptr, type value, ...);
// 将 0 写入到 ptr，并对 ptr 解锁，即 unlock spinlock 语义
void __sync_lock_release(type *ptr, ...)

后面的可扩展参数(...)用来指出哪些变量需要 Memory Barrier，由于目前 GCC 实现的是 Full Barrier，因此可以忽略该参数。

在 GCC 实现了 C++11 之后，上述的 __sync 系列函数便不再推荐使用了，而是基于 C++11 的新原子操作接口，使用 __atomic 作为前缀，对于普通的数学操作函数，其函数接口形式为：

type __atomic_OP_fetch(type* ptr, type val, int memorder);
type __atomic_fetch_OP (type *ptr, type val, int memorder);

其中，OP 表示数学操作符号，类似于上述的 __sync 系列函数。

除此之外，还根据新标准提供了一系列新的接口：

type __atomic_load_n (type *ptr, int memorder);
void __atomic_store_n (type *ptr, type val, int memorder);
type __atomic_exchange_n (type *ptr, type val, int memorder);
bool __atomic_compare_exchange_n (type *ptr, type *expected, type desired, bool weak, int success_memorder, int failure_memorder);
bool __atomic_test_and_set (void *ptr, int memorder);
void __atomic_clear (bool *ptr, int memorder);
void __atomic_thread_fence (int memorder);
bool __atomic_always_lock_free (size_t size, void *ptr);
bool __atomic_is_lock_free (size_t size, void *ptr);

在模板类 AtomicIntegerT 中，只声明了一个成员变量 value_，该变量使用 volatile 关键字进行修饰。

使用 volatile 关键字所修饰的变量表示随时可能会被某个未知的因素进行修改，即表示它是易变的，常用于多线程开发中。如果某个变量被该关键字修饰，那么每次去读取该值时是从对应的地址中去读取，而不会因为编译器发现程序中间没有对该变量进行写入操作而不去内存上读取，这就阻止了编译器对该变量的优化，同时也保证了对特殊地址的稳定访问。

Mutex

线程安全分析

如果使用的是 Clang 编译器，则可以开启线程安全分析(Thread safety annotations)，它是 C++ 语言的扩展，警告代码中潜在的竞争条件。它是完全静态的，没有运行时的开销。例如，如果变量 foo 由锁变量 mu 保护，那么每当一段代码在 foo 没有被锁定的情况下读取或写入时，分析就会发出警告。

#include "mutex.h"

class BankAccount {
private:
  Mutex mu;
  int   balance GUARDED_BY(mu);

  void depositImpl(int amount) {
    balance += amount;       // 警告！不能在没有加锁的情况下写入 balance 变量
  }

  void withdrawImpl(int amount) REQUIRES(mu) {
    balance -= amount;       // 正确！调用者必须加锁
  }

public:
  void withdraw(int amount) {
    mu.Lock();
    withdrawImpl(amount);    // 正确！我们加了锁
  }                          // 警告！并未解锁

  void transferFrom(BankAccount& b, int amount) {
    mu.Lock();
    b.withdrawImpl(amount);  // 警告！调用时需要对 b.mu 进行加锁
    depositImpl(amount);     // 正确，该函数没有需求
    mu.Unlock();
  }

在上述例子中，GUARDED_BY 属性声明线程必须先给 mu 加锁才能读取或写入 balance 变量，从而确保递增和递减操作是原子的。同样，REQUIRES 声明调用线程必须在调用之前加锁。因为调用者被假定为已经加锁，所以在方法的主体内修改变量 balance 是安全的。

要开启线程安全分析，只需使用 -Wthread-safety 标志进行编译即可，例如：

clang -c -Wthread-safety example.cpp

除此之外，还包括：

-Wthread-safety-attributes: 线程安全属性的语义检查
-Wthread-safety-analysis: 核心分析
-Wthread-safety-precise: 要求互斥体表达式精确匹配，对于具有大量别名的代码，可以禁用此警告
-Wthread-safety-reference: 检查受保护成员何时通过引用传递

GUARDED_BY

GUARDED_BY 是数据成员的一个属性，它声明数据成员受给定能力的保护。对数据的读操作需要共享访问，而写操作需要独占访问。PT_GUARDED_BY 类似，但旨在用于指针和智能指针。数据成员本身没有限制，但它指向的数据受给定能力的保护。

Mutex mu;
int *p1             GUARDED_BY(mu);
int *p2             PT_GUARDED_BY(mu);
unique_ptr<int> p3  PT_GUARDED_BY(mu);

void test() {
  p1 = 0;             // Warning!

  *p2 = 42;           // Warning!
  p2 = new int;       // OK.

  *p3 = 42;           // Warning!
  p3.reset(new int);  // OK.
}

REQUIRES

REQUIRES 是函数或方法的一个属性，它声明调用线程必须具有对给定功能的独占访问权限。可以指定不止一种能力。功能必须在进入功能时保留，并且在退出时仍必须保留。REQUIRES_SHARED 类似，但只需要共享访问。

Mutex mu1, mu2;
int a GUARDED_BY(mu1);
int b GUARDED_BY(mu2);

void foo() REQUIRES(mu1, mu2) {
  a = 0;
  b = 0;
}

void test() {
  mu1.Lock();
  foo();         // Warning!  Requires mu2.
  mu1.Unlock();
}

之前称作 EXCLUSIVE_LOCKS_REQUIRED 和 SHARED_LOCKS_REQUIRED

ACQUIRE、RELEASE

ACQUIRE 和 ACQUIRE_SHARED 是函数或方法的属性，声明该函数获得了一项能力，但不释放它。给定的能力不得在进入时保留，而将在退出时保留。

RELEASE、ACQUIRE_SHARED 和 RELEASE_GENERIC 声明该函数释放给定的能力。该功能必须在进入时保留，退出时将不再保留。

Mutex mu;
MyClass myObject GUARDED_BY(mu);

void lockAndInit() ACQUIRE(mu) {
  mu.Lock();
  myObject.init();
}

void cleanupAndUnlock() RELEASE(mu) {
  myObject.cleanup();
}                          // Warning!  Need to unlock mu.

void test() {
  lockAndInit();
  myObject.doSomething();
  cleanupAndUnlock();
  myObject.doSomething();  // Warning, mu is not locked.
}

如果没有参数传递给 ACQUIRE 和 RELEASE，那么参数被假定为 this，并且分析不会检查函数的主体。此模式旨在将加锁细节隐藏在抽象接口后面的类使用。例如：

template <class T>
class CAPABILITY("mutex") Container {
private:
  Mutex mu;
  T* data;

public:
  // Hide mu from public interface.
  void Lock()   ACQUIRE() { mu.Lock(); }
  void Unlock() RELEASE() { mu.Unlock(); }

  T& getElem(int i) { return data[i]; }
};

void test() {
  Container<int> c;
  c.Lock();
  int i = c.getElem(0);
  c.Unlock();
}

之前称作 EXCLUSE_LOCK_FUNCTION、SHARED_LOCK_FUNCTION、UNLOCK_FUNCTION

EXCLUDES

EXCLUDES 是函数或方法的一个属性，它声明调用者不能拥有给定的能力。此关键字用于防止死锁，许多互斥锁实现是不可重入的，因为如何函数第二次获取互斥锁，就会发生死锁。

Mutex mu;
int a GUARDED_BY(mu);

void clear() EXCLUDES(mu) {
  mu.Lock();
  a = 0;
  mu.Unlock();
}

void reset() {
  mu.Lock();
  clear();     // Warning!  Caller cannot hold 'mu'.
  mu.Unlock();
}

与 REQUIRES 不同的是，EXCLUDES 是可选的。如果缺少属性，分析不会发出警告，这在某些情况下可能会导致误报。

之前称作 LOCKS_EXCLUDED

NO_THREAD_SAFETY_ANALYSIS

NO_THREAD_SAFETY_ANALYIS 是函数或方法的一个属性，它关闭该方法的线程安全检查。

class Counter {
  Mutex mu;
  int a GUARDED_BY(mu);

  void unsafeIncrement() NO_THREAD_SAFETY_ANALYSIS { a++; }
};

与其他属性不同，NO_THREAD_SAFETY_ANALYSIS 不是函数接口的一部分，因此应放在函数定义(.cc 文件或 .cpp 文件)而不是函数声明中(头文件)。

RETURN_CAPABILITY

RETURN_CAPABILITY 是函数或方法的属性，它声明函数返回对给定功能的引用。它用于注释返回互斥锁的 getter() 方法。

class MyClass {
private:
  Mutex mu;
  int a GUARDED_BY(mu);

public:
  Mutex* getMu() RETURN_CAPABILITY(mu) { return &mu; }

  // analysis knows that getMu() == mu
  void clear() REQUIRES(getMu()) { a = 0; }
};

之前称作：LOCK_RETURNED

ACQUIRED

ACQUIRED_BEFORE 和 ACQUIRED_AFTER 是成员声明的属性，特别是互斥锁或其他功能的声明。这些声明强制执行必须获取互斥锁的特定顺序，以防止死锁。

Mutex m1;
Mutex m2 ACQUIRED_AFTER(m1);

// Alternative declaration
// Mutex m2;
// Mutex m1 ACQUIRED_BEFORE(m2);

void foo() {
  m2.Lock();
  m1.Lock();  // Warning!  m2 must be acquired after m1.
  m1.Unlock();
  m2.Unlock();
}

CAPABILITY

CAPABILITY 是类的一个属性，它指定类的对象可以用作能力，string 参数指定错误消息中的能力类型，例如 “mutex”。

template <class T>
class CAPABILITY("mutex") Container {
private:
  Mutex mu;
  T* data;

public:
  // Hide mu from public interface.
  void Lock()   ACQUIRE() { mu.Lock(); }
  void Unlock() RELEASE() { mu.Unlock(); }

  T& getElem(int i) { return data[i]; }
};

void test() {
  Container<int> c;
  c.Lock();
  int i = c.getElem(0);
  c.Unlock();
}

SOCPED_CAPABILITY

SCOPED_CAPABILITY 是实现 RAII 样式锁定的类的属性，其中在构造函数中获取能力，并在析构函数中释放。此类类需要特殊处理，因为构造函数和析构函数通过不同的名称引用能力。

SCOPED_CAPABILITY 被视为在构造时隐式获取并在销毁时释放的能力。它们与构造函数的线程安全属性中命名的一组常规功能相关联。其他成员函数上的获取类型属性被视为应用于该组的关联功能，同时 RELEASE 意味着函数以任何模式释放所有关联功能。

以前称作 SCOPED_LOCKABLE

TRY_ACQUIRE

这些是试图获取给定能力的函数或方法的属性，并返回一个指示成功或失败的布尔值。第一个参数必须是 true 或 false，以指示哪个返回值指示成功，其余参数的解释与 ACQUIRE 相同。

Mutex mu;
int a GUARDED_BY(mu);

void foo() {
  bool success = mu.TryLock();
  a = 0;         // Warning, mu is not locked.
  if (success) {
    a = 0;       // Ok.
    mu.Unlock();
  } else {
    a = 0;       // Warning, mu is not locked.
  }
}

以前称作 EXCLUSIVE_TRYLOCK_FUNCTION、SHARED_TRYLOCK_FUNCTION

ASSERT_CAPABILITY

这些是断言调用线程已经拥有给定能力的函数或方法的属性，例如通过执行运行时测试并在未用有能力时终止。此注释的存在导致分析假定在调用注释函数后保留该功能。

以前称作 ASSERT_EXCLUSIVE_LOCK、ASSERT_SHARED_LOCK

返回值的检查

源码如下：

#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 // !NDEBUG
#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

其中，__attribute__ ((__noreturn__)) 表示告诉编译器该函数不会返回，用来抑制关于未到达代码路径的错误。比如 C 库函数中的 abort() 和 exit()：

extern void exit(int)   __attribute__ ((__noreturn__));
extern void abort(void) __attribute__ ((__noreturn__));

其中的 __typeof__ 相当于 typeof()，而 __typeof__(ret) errnum 就是将 errnum 的数据类型定义为和 ret 相同的数据类型。

其中的 __BEGIN_DECLS 和 __END_DECLS 是一组宏，其定义如下：

#if defined(__cplusplus)
    #define __BEGIN_DECLS   extern "C" {
    #define __END_DECLS     }
#else
    #define __BEGIN_DECLS
    #define __END_DECLS

也就是说，这两宏之间的部分相当于声明了 extern "C"，即将该部分代码作为 C 语言代码进行使用。

MutexLock

MutexLock 用于封装临界区，使用 RAII 手法封装互斥器的创建与销毁。

在 MutexLock 的定义中，有两个成员变量，分别是 mutex_ 和 holder_，前者为要操作的互斥锁，后者代表当前持有互斥锁的线程 ID，如果该值大于 0，则表示互斥锁正在被占用，否则表示锁可用。

MutexLock()
  : holder_(0) {
  MCHECK(pthread_mutex_init(&mutex_, NULL));
}

~MutexLock() {
  assert(holder_ == 0);
  MCHECK(pthread_mutex_destroy(&mutex_));
}

MutexLock 在构造函数中初始化锁并将 holder_ 变量初始化为 0。在析构函数中先判断当前互斥锁是否被占用，如果没有被占用则销毁锁。

void unassignHolder() {
  holder_ = 0;
}

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

其中，unassignHolder() 用来释放线程句柄，assignHolder() 则用来分配线程句柄。

void lock() ACQUIRE() {
  MCHECK(pthread_mutex_lock(&mutex_));
  assignHolder();
}

void unlock() RELEASE() {
  unassignHolder();
  MCHECK(pthread_mutex_unlock(&mutex_));
}

lock() 用来加锁，其内部先对互斥锁加锁，然后分配线程句柄，以保证线程安全。而 unlock() 则用来解锁，先释放线程句柄，然后对互斥锁解锁。

MutexLock 将 Condition 设置为其友元类，同时声明内嵌类 UnassignGuard 供 Condition 使用。

class UnassignGuard : noncopyable {
public:
  explicit UnassignGuard(MutexLock& owner)
    : owner_(owner) {
    owner_.unassignHolder();
  }

  ~UnassignGuard() {
    owner_.assignHolder();
  }
private:
  MutexLock& owner_;
};

其中，UnassignGuard 在构造时释放线程句柄，而在析构时则分配线程句柄。

MutexLockGuard

MutexLockGuard 用于封装临界区的进入和退出，即加锁、解锁操作。其一般为栈上对象，作用域刚好等于临界区域。

同样地，MutexLockGuard 也使用了 RAII 手法，在构造函数中加锁，在析构函数中解锁。

explicit MutexLockGuard(MutexLock& mutex) ACQUIRE(mutex)
  : mutex_(mutex) {
  mutex_.lock();
}

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

注意，在 MutexLockGuard 类内部只有一个成员变量：

MutexLock& mutex_;

这里使用 “引用” 的原因主要有两点：

对构造函数中传入的 MutexLock 对象本身进行加锁操作；
MutexLockGuard 销毁时并不会销毁原先的互斥锁；

在类外，定义了如下宏：

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

其目的是为了防止生成一个无名的临时 MutexLockGuard 对象，因为该临时对象没有持久的作用域，不能长时间持有 MutexLock 对象。

Condition

Condition 类简单的对条件变量进行了封装。其成员变量有两个，分别是 mutex_ 和 pcond_，前者用来绑定互斥锁，后者用来作为条件变量。

同样的，Condition 也使用了 RAII 机制：

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

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

Condition 的阻塞等待如下：

void wait() {
  MutexLock::UnassignGuard ug(mutex_);
  MCHECK(pthread_cond_wait(&pcond_, mutex_.getPthreadMutex()));
}

bool waitForSeconds(double seconds) {
  struct timespec abstime;
  clock_gettime(CLOCK_REALTIME, &abstime);

  const int64_t kNanoSecondsPerSecond = 1000000000;
  int64_t nanoseconds = static_cast<int64_t>(seconds * kNanoSecondsPerSecond);

  abstime.tv_sec += static_cast<time_t>((abstime.tv_nsec + nanoseconds) / kNanoSecondsPerSecond);
  abstime.tv_nsec = static_cast<long>((abstime.tv_nsec + nanoseconds) % kNanoSecondsPerSecond);

  MutexLock::UnassignGuard ug(mutex_);
  return ETIMEDOUT == pthread_cond_timedwait(&pcond_, mutex_.getPthreadMutex(), &abstime);
}

创建 UnassignGuard 对象的目的是为了和 pthread_cond_wait() 达到同样的效果，该函数通常有以下三步操作：

释放 Mutex；
阻塞等待；
当被唤醒时，重新获得 Mutex 并返回；

所以在此处构建 UnassignGuard 对象以释放线程句柄，而在 pthread_cond_wait() 内部则会释放互斥锁，阻塞等待互斥锁可用。当被唤醒时，pthread_cond_wait() 函数返回并对互斥锁上锁，此时析构 UnassignGuard 对象以分配线程句柄，重新获得互斥锁。此过程类似于 MutexLock::lock() 和 MutexLock::unlock() 操作。

唤醒阻塞线程的方法如下：

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

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

notify() 用于唤醒某个在条件变量上阻塞等待的另一个线程，notifyAll() 用于唤醒在当前条件变量上阻塞等待的所有线程。

注意，如果一个类要包含 MutexLock 和 Condition，要注意它们的声明顺序和初始化顺序，MutexLock 应先于 Condition 构造，并作为后者的构造参数。

CountDownLatch

CountDonwLatch 是一个用于倒计时的同步辅助类，比如在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。既可以用于所有子线程等待主线程完成事务，也可以用于主线程等待子线程完成初始化工作。

该类与上方几个类的成员变量有所不同，并未采用 “引用” 的方式，说明互斥锁、条件变量仅为内部所有。其成员变量有如下三个：mutex_、condition_ 和 count_。mutex_ 主要用于临界区的代码保护，condition_ 主要配合 mutex_ 使用，而 count_ 则用于表示计数器，当计数器为 0 时，通知所有阻塞等待的线程。

注意，mutex_ 变量被 mutable 关键字所修饰，该关键字用于表明被修饰变量在常成员函数内部也可以被修改。

int CountDownLatch::getCount() const {
  MutexLockGuard lock(mutex_);
  return count_;
}

由于在常成员函数 getCount() 内部需要对互斥锁加锁和解锁，所以需要 mutable 关键字来进行修饰。

向阻塞线程发送通知的线程调用如下函数，不断减少 count_ 的数量，直到其为 0，则通知所有阻塞线程。

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

而被阻塞线程则调用如下函数以进行等待操作，在 count_ 数量减为 0 之前，一直处于等待唤醒状态。

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

注意：所有线程对象都需要调用同一个 CountDownLatch 实例，公用同一个 MutexLock 和 Condition。

标签：__,Muduo,void,mu,原语,同步,线程,type,ptr
From： https://www.cnblogs.com/tuilk/p/16792088.html