§ 1 线程安全的对象生命期管理

  • 对象的生与死不能由对象自身拥有的mutex(互斥器)来保护。如何避免对象析构时可能存在的race condition(竞态条件)是C++多线程编程面临的基本问题。

  • 当一个对象被多个线程同时看到,那么对象的销毁时机就会模糊不清,可能出现多种的race condition(竞态条件):

    • 即将析构一个对象时,从何而知此刻是否有别的对象正在执行该对象的成员函数?
    • 如何保证在执行成员函数期间,对象不会被另一个线程所析构?
    • 在调用某个对象的成员函数之前,如何得之这个对象还活着?它的析构函数是否会碰巧执行到一半?
  • 一个线程安全的class应当满足下面三个条件:

    • 多个线程同时访问时,其表现出正确的行为。
    • 无论操作系统如何调度这些线程,无论这些线程的执行顺序如何交织(interleaving)。
    • 调用端代码无需额外的同步或其它协调动作。
  • 按照以上定义,C++标准库中的大多数类都不是线程安全的,包括std:string、std::vector、std::map等,这些class通常需要在外部加锁才能供多个线程同步访问。

  • 对象构造要做到线程安全,唯一的要求是在构造期间不要泄漏this指针,即:

    • 不要在构造函数中注册任何回调。
    • 不要在构造函数中把this指针传给跨线程的对象。
    • 即使是构造函数最后一行也不行(如果该类是一个基类,基类会先于派生类构造,执行完父类构造函数的最后一句后会接着执行子类的构造函数,这时most-derived class的对象还处于构造中,仍然不安全)。
  • 一个函数如果要锁住相同类型的多个对象,为了始终按照相同的顺序加锁,我们可以比较mutex对象的地址,始终先加锁地址较小的对象。

  • 一个对象在析构的时候,调用它的任何非静态函数都是不安全的。

  • scoped_ptr/shared_ptr/weak_ptr都是值语义,要么是栈上对象,或者是其它对象的直接数据成员,其“计数”操作在主流平台上的实现都是原子操作。

  • 注意shared_ptr本身不是线程安全的,它的引用计数本身是安全无锁的,但对象的读写却不是。如果多个线程要读写同一个shared_ptr,那么需要加锁。

  • 如果对象的析构比较耗时,我们可以用一个单独的线程专门做析构,通过一个BlockingQueue<shared_ptr>把对象的析构都挪到专门的析构线程。

  • 分析可能出现的rece condition不仅是多线程编程的基本功,也是分布式系统设计的基本功,需要反复历练,形成一定的思考范式,并积累一些经验教训,才能少犯错误。

  • 用流水线、生产者消费者、任务队列等这些有规律的机制,最低限度的共享数据。这是我已知的最好的多线程编程的建议了。

§ 2 线程同步精要

  • 并发编程有两种基本模型,一种是message passing(消息传递),另一种是shared memory(共享内存)。在分布式系统中,运行在多台机器的多个进程的并行编程只有一种实用的模型:message passing。

  • 线程同步的四项重要原则:

    • 首要原则是最低限度的共享对象,减少需要同步的场合。
    • 其次是尽量使用高级的并发编程构建,如TaskQueue、Producer-Consumer Queue等等。
    • 最后不得已必须使用底层同步原语(primitives)时,只用非递归(不可重入)的互斥器和条件变量,慎用读写锁,不要用信号量。
    • 除了使用atomic整数之外,不要再自己编写lock-free代码,也不要使用内核级的同步原语,不凭空猜测哪种效率可能会更好。
  • 互斥器(mutex)注意点:

    • 用RAII手法封装mutex的创建、销毁、加锁、解锁操作。
    • 只使用不可重入的mutex(非递归)。
    • 不手工调用lock()和unlock()函数,一切交给栈上的Guard对象的构造和析构函数负责。Guard对象的生命期正好等于临界区。
    • 在每次构造Guard对象的时候,思考栈上已经持有的锁,放置因为加锁顺序所导致的死锁(deadlock)。
    • 不使用跨进程的mutex,进程间通信只使用TCP Sockets。
    • 加锁/解锁在同一个线程里进行,线程a不能去unlock线程b已经锁住的mutex(RAII自动保证)。
  • Linux下mutex的一个封装实例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
// Mutex.h

#ifndef MUTEX_H_
#define MUTEX_H_

#include <pthread.h>

#include "noncopyable.h"

/**
* 原始互斥锁的封装
*/
class MutexLock : private noncopyable
{
public:
MutexLock();
~MutexLock();

/// 加锁函数(仅允许 MutexLockGuard 类调用)
void lock();

/// 解锁函数(仅允许 MutexLockGuard 类调用)
void unlock();

/// 返回原始互斥锁类型的指针(仅允许 Condition 类调用)
pthread_mutex_t *getPthreadMutex();

private:
pthread_mutex_t mutex_;
};

/**
* MutexLock 互斥锁的加/解锁类
*/
class MutexLockGuard : private noncopyable
{
public:
/// 构造函数自动加锁
explicit MutexLockGuard(MutexLock &mutex) : mutex_(mutex)
{
mutex_.lock();
}

/// 脱离作用域的时候自动解锁
~MutexLockGuard()
{
mutex_.unlock();
}

private:
MutexLock &mutex_;
};

/**
* 下面的宏防止出现诸如 MutexLockGuard(mutex) 的定义
* 正规做法是 MutexLockGuard lock(mutex)
*/
#define MutexLockGuard(x) static_assert(false, "missing mutex guarg var name!")

#endif // MUTEX_H_
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
// MutexLock.cpp
#include "MutexLock.h"

MutexLock::MutexLock()
{
pthread_mutexattr_t attr;

pthread_mutexattr_init(&attr);
/// 初始化互斥锁类型为 PTHREAD_MUTEX_NORMAL
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_NORMAL);
pthread_mutex_init(&mutex_, &attr);
pthread_mutexattr_destroy(&attr);
}

MutexLock::~MutexLock()
{
pthread_mutex_destroy(&mutex_);
}

void MutexLock::lock()
{
pthread_mutex_lock(&mutex_);
}

void MutexLock::unlock()
{
pthread_mutex_unlock(&mutex_);
}

pthread_mutex_t *MutexLock::getPthreadMutex()
{
return &mutex_;
}
  • 条件变量(condition variable)用于一个或者多个线程等待某个布尔表达式为真,即等待别的线程“唤醒”它,条件变量的学名叫管程(monitor)。对于wait端:

    • 条件变量必须与mutex一起使用,该布尔表达式的读写需受此mutex保护。
    • 在mutex已经上锁的时候才能调用wait()。
    • 把判断布尔条件和wait()放到while循环里(避免虚假唤醒(suprious wakeup))。
    • 对于signal/broadcast端
    • 不一定要再mutex已经上锁的情况下调用signal(理论上)。
    • 在signal之前一定要修改布尔表达式。修改布尔表达式的时候需用mutex保护。
    • 注意区分signal和broadcast。
  • Linux下condition封装的一个实例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
// Condition.h

#ifndef CONDITION_H_
#define CONDITION_H_

#include <pthread.h>

#include "noncopyable.h"
#include "MutexLock.h"

/**
* 条件变量的封装类
*/
class Condition : private noncopyable
{
public:
/// 构造函数
explicit Condition(MutexLock &mutex) : mutex_(mutex)
{
pthread_cond_init(&pcond_, NULL);
}

/// 析构函数
~Condition()
{
pthread_cond_destroy(&pcond_);
}

/// 等待该条件变量
void wait()
{
pthread_cond_wait(&pcond_, mutex_.getPthreadMutex);
}

/// 通知一个等待该条件变量的线程
void notify()
{
pthread_cond_signal(&pcond_);
}

/// 通知所有等待该条件变量的线程
void notifyAll()
{
pthread_cond_broadcast(&pcond_);
}

private:
MutexLock &mutex_;
pthread_cond_t pcond_;
};

#endif // CONDITION_H_
  • 真正影响性能的不是锁,而是锁争用(lock contention)。

  • 必须用单线程的场合:

    • 程序有可能会fork(2);
    • 限制程序的CPU使用率。
  • 多线程的适用场景是:提高响应速度,让IO和“计算”相互重叠,降低latency。虽然多线程不能提高绝对性能,但能提高平均响应性能。

赏杯咖啡鼓励下~