多进程之间存在互斥和合作关系时需要同步控制以保证执行结果的逻辑正确, 多个线程共享资源和合作时, 也同样存在同步的问题, 5.5.1 引入问题时, 我们就已经看到了线程不同步造成的加和错误。 同进程同步类似, 线程也需要有处理同步的机制, 如果一个进程中的每个线程同时去修改全局内存的数据, 而操作函数没有使用同步机制防止数据在共享使用中被破坏, 程序执行就不是线程安全(Thread-safeness)的。

Pthreads API 中的函数有基础的线程管理类函数, 包括线程创建(Creating)、 分离(Detaching)、 连接(Joining), 以及用于设置和查询线程属性(可连接, 调度属性等)的函数等。Pthreads API 还提供了用于线程同步的函数。 互斥量(Mutexes)函数提供了创建、 销毁、 锁定和解锁互斥量的功能, 还包括了一些用于设定或修改互斥量属性的函数。 而条件变量(Condition Variables)类函数通过设计条件变量, 结合共享的互斥量实现线程间同步控制,这类函数包括条件变量的创建、 销毁、 等待和释放等。

互斥锁的简单性是其优点也是其缺点, 由于它只有两种状态: 锁定和非锁定, 在处理复杂同步问题时, 尤其顺序问题时, 就容易出现反复试探的问题。 而条件变量支持阻塞等待和唤醒操作, 通过允许线程阻塞和等待另一个线程将其唤醒的方法弥补了互斥锁的不足, 它常和互斥锁一起使用, 主要流程是:

为了访问资源执行后续代码, 先加互斥锁, 然后判断条件; 如果条件不满时, 线程解开相应的互斥锁并阻塞等待, 一旦其他的某个线程改变了条件变量, 将唤醒一个或多个正因此条件变量阻塞的线程; 被唤醒的线程将重新锁定互斥锁并重新测试条件是否满足, 若满足就能顺利执行后续对资源操作。

信号量本质上是“一个计数器+一个互斥量+一个等待队列”, 其计数器能进一步体现资源的量, 其适用面广, 适用于各种进程或线程的同步控制; 而条件变量适用于线程同步, 相比信号量而言的好处也就是更轻便、 灵活, 它还有能够一次唤醒所有阻塞对象的特殊操作。 所以在不同场合, 可以根据问题的需求选择不同的同步控制手段。

1. 互斥锁实现线程互斥

(1) 互斥锁的初始化与销毁。

静态初始化方法需要定义常数PTHREAD_MUTEX_INITIALIZER 赋值给一pthread_mutex_t 变量:

这种定义是普通锁, 同一线程可重复加锁, 解锁一次释放锁, 先等待锁的进程先获得锁。

静态定义的值根据赋值, 有不同类型的锁: 嵌套锁、 纠错锁、 自适应锁。

动态地创建互斥量时, 在申请内存(malloc)之后, 用pthread_mutex_init()初始化互斥对象, 销毁时使用pthread_mutex_ destroy() , 函数原型如下。

pthread_mutex_init 接受一个指针作为参数以初始化为互斥对象, 该指针指向一块已分配好的内存区。 第二个参数, 可以接受一个可选的pthread_mutexattr_t 指针, 用来设置各种互斥对象属性, 但通常并不需要这些属性, 可指定为NULL。

pthread_mutex_destroy()接受一个指向pthread_mutext_t 的指针作为参数, 可释放创建互斥对象时分配给它的任何资源。 注意, pthread_mutex_destroy()不会释放用来存储pthread_mutex_t 的内存。

pthread_mutex_init()和pthread_mutex_destroy()成功时都返回零。

(2) 加/解锁操作。

对共享资源访问时要对互斥量进行加锁, 如果互斥量已经上了锁, 执行加锁操作的线程会阻塞, 直到占用互斥量的对象在完成对共享资源的访问后对互斥量进行解锁。

相关函数原型如下。

函数执行成功均返回0, 出错则返回错误编号。 注意trylock 函数, 该函数是非阻塞调用模式, 其执行判断互斥量被锁住时, 函数不会阻塞等待而直接返回EBUSY, 表示共享资源处于忙状态。

【例5-15】 两个线程共享进程里的a、 b 变量, 要求ThreadB 负责输出ThreadA 对a、 b做运算后的结果, 要求ThreadB 在ThreadA 运算过程中不能访问变量做输出。

其执行结果如下。

两个程序并发可能会有各种执行顺序, 靠它动态运行去发现问题比较麻烦, 所以为了方便地测试可能的执行顺序, 我们人工调整A 线程里sleep 的秒数模拟AB 不同顺序情况下的运行效果。 先注释掉B 线程里的互斥量操作代码(代码中标注有#的语句), 体验没有控制下的混乱。

A 线程里设置sleep(2), 因为A 线程睡眠时间比B 短, 在B 睡眠时, A 已经能完成规定的计算, B 输出结果是期望的200。

A 线程里设置sleep(5), 因为A 线程睡眠时间比B 长, 所以在计算中途, B 就获得了变量做输出, 这和我们要求A 的计算完毕前B 不能操作相反, 输出结果为150。

这个过程模拟了如果两线程以不同的速度执行, 完全有可能得到不同的执行结果。 这就是所谓的程序执行没有可再现的结果, 也和我们开始对程序功能的要求不一致。 而且可以看出来, 即使A 线程里设置了互斥量, B 线程里注释掉该语句, 没有和A 争抢信号的对象, 该信号设置上也没有意义。

最后, 去掉B 线程里的信号量操作语句前注释符, 让语句起效, 重新编译执行, 可多次修改sleep 的时间测试运行, 会发现无论他们睡眠时间是什么情况, B 输出的始终是按规定逻辑等A 完整的计算完后的结果。

2. 条件变量实现线程同步

(1) 条件变量的初始化与销毁。

静态初始化方法需要定义常数PTHREAD_COND_INITIALIZER 赋值给一pthread_cond_t 变量:

pthread_ cond_ init 接受一个指针作为参数以初始化, 该指针指向一块已分配好的内存区。 第二个参数, 可以接受一个可选的指针, 用来设置各种互斥对象属性, 一般可指定为NULL。

pthread_cond_destroy()只有在没有线程在该条件变量上等待的时候才能注销条件变量, 否则返回EBUSY。 因为Linux 实现的条件变量没有分配什么资源, 所以注销动作只包括检查是否有等待线程。 注意new 开辟的pthread_ cond_ t 在调用pthread_ cond_ destroy()后要调用delete 或者free 销毁掉。(www.daowen.com)

(2) 阻塞/唤醒操作。

某种条件不满足时阻塞线程, 函数原型如下。

cond 是管理阻塞队列的指针, restrict mutex 是传给条件变量操作的互斥锁。

一般条件变量是与互斥量是配合使用的, 先通过了互斥量的加锁再到此函数的执行,基于对某种条件的判断决定是否阻塞进程。 阻塞进程时, 该函数执行将释放互斥锁, 并将当前调用者阻塞在条件变量cond 上。 调用成功时返回0, 否则返回错误编号以指示错误。

需要注意, 如果调用执行线程被阻塞, 当其被唤醒时, restrict mutex 会被重新加锁,以与进入pthread_cond_wait()前的加锁动作对应, 这也保证了线程被唤醒后, 互斥量会保护其后续对共享资源的操作仍是互斥的。

唤醒等待条件变量的线程, 函数原型如下。

基于对某种条件的判断, 激活一个等待在cond 条件变量上的线程, 存在多个等待线程时按入队顺序激活其中一个; 另外还有激活所有等待线程的调用pthread_cond_broadcast(pthread_cond_t∗cond)。 函数调用成功时返回0, 否则返回错误号以指示错误。

【例5-16】 一个生产者和一个消费者同步实例如下。

其执行结果如下。

本例中生产者不断放入, 消费者没有产品时不能取出产品, 是一个有明确顺序关系的同步过程。

可以看到生产者和消费者线程一运行就争抢互斥信号mutex, 对于生产者来说, 直接使计数变量增长以模拟生产操作。 如果此时无人等待在条件变量上, 其条件变量的signal释放操作实际上不需要做什么工作。

而对消费者来说, 一定要注意pthread_ cond_ wait(&cond, &mutex)和前一句count<0的判断密切相关, 如果判断成立, 此步会放弃它自己对mutex 互斥锁的占用, 此线程会释放互斥锁并转阻塞操作。

为了测试同步情况, 我们在代码中特意将consumer 线程建立放前面, 同时让消费者sleep 的时间比producer 的sleep 时间长, 通过测试, 我们看到了两者运行符合逻辑要求,确实实现了顺序控制。 有互斥锁的保障, count 的计数也没错。

3. 信号量控制线程同步

POSIX 标准指定了信号量的接口, 它不是Pthread 的一部分, 但是大多数实现Pthread的UNIX 也提供信号量编程接口。 信号量本质上是一个非负的整数计数器, 它被用来控制对共享资源的访问。 当信号量值大于0 时, 才能使用共享资源, 通过sem_wait()减少信号量值, 控制程序是否阻塞。 函数sem_trywait()是函数sem_wait()的非阻塞版本。 当共享资源被释放时, 调用函数sem_post()增加信号量值。 信号量有关的函数都在头文件/usr/include/semaphore.h 中定义。

信号量的数据类型为结构sem_t, 本质上是一个长整型的数。

(1) 信号量创建及初始化。

sem 为指向信号量结构的一个指针; pshared 大于0 表示信号量在进程间共享, 为0 则表示仅当前进程的所有线程共享; value 设置信号量的初始值。

函数sem_destroy(sem_t∗sem)用来销毁信号量sem, 调用成功则返回0, 失败返回-1。

(2) 信号量的申请。

函数sem_ wait( sem_ t ∗sem ), sem_ wait 是一个阻塞的原子操作函数, 测试所指定信号量的值, 若sem 参数的值大于0, 则该信号量值减去1 并立即返回; 若值等于0, 阻塞直到sem 的值被其他线程影响大于0 时立即减去1, 然后返回。

函数sem_trywait ( sem_t ∗sem )是函数sem_wait()的非阻塞版本, 它尝试获取sem 的值, 如果值为0, 不阻塞而是直接返回一个错误EAGAIN; 否则直接将信号量sem 的值减1。

(3) 信号量的释放。

该函数用来增加信号量的值。 当有线程阻塞在这个信号量上时, 调用这个函数会使其中的一个线程不再阻塞, 选择唤醒谁是由线程的调度策略决定。

(4) sem_ getvalue。

获取信号量sem 的当前值, 把该值保存在sval, 若有1 个或者多个线程正在调用sem_wait 阻塞在该信号量上, 该函数返回阻塞在该信号量上进程或线程个数。

【例5-17】 与5.4.3【例5-11】不同的是, 该例的线程顺序是利用基本线程管理函数控制父子进程间的顺序问题, 而本例是要通过信号量控制任意两个线程的合作。 主线程产生了两个线程, 让他们一个模拟生产放入, 一个模拟取水果吃。

其执行结果如下。

调整两线程中sleep 语句的睡眠时间, 可以测试不同执行顺序同步控制的效果。