unix环境高级编程(阅读笔记)----线程同步

多个线程共享相同的内存时，需要确保每个线程看到一致的数据视图。

1.互斥量

可以通过使用pthread的互斥接口保护数据，确保同一时间只有一个线程访问数据，互斥量（mutex）从本质上说是一把
锁，在访问共享资源前对互斥量进行枷锁，在访问完成后释放互斥量上的锁。对互斥量进行加锁后，任何其他试图再次对互斥
量进行加锁的线程将被阻塞知道当前线程释放该互斥锁。如果释放互斥锁时有多个线程阻塞，所有在该互斥锁上阻塞线程都会
变成可运行状态，第一个变为运行状态的线程可以对互斥量进行加锁，其他线程将会看到互斥锁依然被锁住，只有回去再次等
待它重新变成可用。在这种方式下，每次只有一个线程可以向前执行。
互斥变量用pthread_mutex_t数据类型来表示，在使用互斥变量以前，必须首先对它进行初始化，可以把它置为常量
PTHREAD_MUTEX_INITIALIZER(只对静态分配的互斥量)，也可以通过调用pthread_mutex_init函数进行初始化。如果动态地
分配互斥量（例如通过调用malloc），那么在释放内存前需要调用pthread_mutex_destroy。

[cpp] view
plain copy







#include<pthread.h>

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);

int pthread_mutex_destroy(pthread_mutex_t * mutex);

//成功返回0，否则返回错误编号

使用默认的属性初始化互斥量，只要将attr设置为NULL。

对互斥量进行加锁，需要调用pthread_mutex_lock，如果互斥量已经上锁，调用线程将阻塞直到互斥量被解锁。对互斥量
解锁，需要调用pthread_mutex_unlock。

[cpp] view
plain copy







#include <pthread.h>

int pthread_mutex_lock(pthread_mutex_t *mutex);

int pthread_mutex_trylock(pthread_mutex)_t *mutex);

int pthread_mutex_unlock(pthread_mutex_t *mutex);

//如果成功返回0，否则返回错误编号。

如果线程不希望被阻塞，它可以使用pthread_mutex_trylock尝试对互斥量进行加锁。如果调用pthread_mutex_trylock时
互斥量处于未锁住状态，那么pthread_mutex_trylock将锁住互斥量，不会出现并返回0，否则pthread_mutex_trylock就会失败，
不能锁住互斥量，而返回EBUSY。

实践：
在没有使用互斥锁时：

[cpp] view
plain copy







#include <stdio.h>

#include <pthread.h>

#include <malloc.h>

#include <string.h>

void* th_func(void* arg){

int i;

for(i=0; i<5; i++){

printf("1\n");

sleep(1);

printf("2\n");

}

}

int main(void){

int ret;

pthread_t tid1,tid2;

ret = pthread_create(&tid1, NULL, th_func, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

return -1;

}

ret = pthread_detach(tid1);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

return -1;

}

ret = pthread_create(&tid2, NULL, th_func, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

return -1;

}

ret = pthread_detach(tid2);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

return -1;

}

sleep(15);

return 0;

}

程序输出为：
1

1

2

1

2

1

2

1

2

1

2

1

2

1

2

1

2

1

2

2

使用了互斥锁之后：

[cpp] view
plain copy







#include <stdio.h>

#include <pthread.h>

#include <malloc.h>

#include <string.h>

pthread_mutex_t *mutex;

void* th_func(void* arg){

int i;

for(i=0; i<5; i++){

pthread_mutex_lock(mutex);

printf("1\n");

sleep(1);

printf("2\n");

pthread_mutex_unlock(mutex);

}

}

int main(void){

int ret,result = 0;

pthread_t tid1,tid2;

mutex = (pthread_mutex_t*)malloc(sizeof(pthread_mutex_t));

if(mutex == NULL){

perror("malloc");

result = -1;

goto FINALLY;

}

pthread_mutex_init(mutex,NULL);

ret = pthread_create(&tid1, NULL, th_func, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid1);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_create(&tid2, NULL, th_func, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid2);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

sleep(15);

FINALLY:

if(mutex != NULL){

pthread_mutex_destroy(mutex);

free(mutex);

}

return result;

}

程序输出为：
1

2

1

2

1

2

1

2

1

2

1

2

1

2

1

2

1

2

1

2

2.避免死锁

如果线程试图对同一个互斥量加锁两次，那么它自身就会陷入死锁状态，使用互斥量时，还有其他更不明显的方式也能产生死锁，
例如，程序中使用多个互斥量，如果允许一个线程一直占有一个互斥量，并且试图锁住第二个互斥量时处于阻塞状态，但是拥有第二
个互斥量的线程也在试图锁住第一个互斥量，这时就会发生死锁。因为两个线程都在相互请求另一个线程拥有的资源，所以这两个线程
都无法向前运行，于是就产生死锁。
可以通过控制互斥量加锁的顺序来避免死锁的发生。例如，假设需要对两个互斥量A和B同时加锁，如果所有线程总是在对互斥量B
加锁之前锁住A，那么使用这两个互斥量不会产生死锁，类似地，如果所有的此案成总是在锁住互斥量A之前锁住B，那么也不会发生死
锁。

3.读写锁

读写锁与互斥量类似，不过读写锁允许更高的并行性。互斥量要么是锁住状态要么不加锁状态，而且一次只有一个线程可以对其
加锁。读写锁可以由三种状态：读模式下加锁状态，写模式下加锁状态，不加锁状态。一次只有一个线程可以占有写模式的读写锁，
但是多个线程可以同时占有读模式的读写锁。
当读写锁是写加锁状态时，在这个锁被解锁之前，所有试图对这个锁加锁的线程都会被阻塞。当读写锁在读加锁状态时，所有试图
以读模式对它进行加锁的线程都可以得到访问权，但是如果线程希望以写模式对此锁进行加锁，他必须阻塞直到所有的线程释放读锁。
当读写锁处理读模式锁住状态时，如果有另外的线程试图以写模式加锁，读写锁通常会阻塞随后的读模式锁请求，这样可以避免读模式
锁长期占用，而等待的些模式请求一直得不到满足。
读写锁非常适用于对数据结构读的次数远大于写的情况。
通过调用pthread_rwlock_init进程初始化，如果希望读写锁有默认的属性，可以传一个空指针给attr。
在释放读写锁占用的内存之前，需要调用pthread_rwlock_destroy做清理工作。

[cpp] view
plain copy







#include <pthread.h>

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);

int pthread_rwlock_destory(pthread_rwlock_t *rwlock);

//若成功返回0，否则返回错误编号。

在读模式下锁住读写锁，需要调用pthread_rwlock_rdlock，要在写模式下锁定读写锁，需要调用pthread_rwlock_wrlock，不管以
何种方式锁住读写锁，都可以调用pthread_rwlock_unlock进行解锁。

[cpp] view
plain copy







#include<pthread.h>

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);

int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);

int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

//成功则返回0，否则返回错误编号。

Single UNIX Specification同样定义了有条件的读写锁原语版本。

[cpp] view
plain copy







#include <pthread.h>

int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

int pthread_rwlock_tryrwlock(pthread_rwlock_t *rwlock);

//成功返回0，否则返回错误编号。

实践：

[cpp] view
plain copy







#include <stdio.h>

#include <pthread.h>

#include <malloc.h>

#include <string.h>

pthread_rwlock_t *rwlock;

int gi=0;

void* th_func1(void* arg){

int i;

for(i=0; i<5; i++){

pthread_rwlock_rdlock(rwlock);

printf("start1:%d\n",gi);

sleep(1);

printf("end1:%d\n",gi);

pthread_rwlock_unlock(rwlock);

sleep(1);

}

}

void* th_func2(void* arg){

int i;

for(i=0; i<5; i++){

pthread_rwlock_rdlock(rwlock);

printf("start2:%d\n",gi);

sleep(1);

printf("end2:%d\n",gi);

pthread_rwlock_unlock(rwlock);

sleep(1);

}

}

void* th_func3(void* arg){

int i;

for(i=0; i<5; i++){

pthread_rwlock_wrlock(rwlock);

gi++;

sleep(1);

pthread_rwlock_unlock(rwlock);

sleep(1);

}

}

int main(void){

int ret,result = 0;

pthread_t tid1,tid2,tid3;

rwlock = (pthread_rwlock_t*)malloc(sizeof(pthread_rwlock_t));

if(rwlock == NULL){

perror("malloc");

goto FINALLY;

}

ret = pthread_create(&tid1, NULL, th_func1, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid1);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_create(&tid2, NULL, th_func2, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid2);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_create(&tid3, NULL, th_func3, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid3);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

sleep(15);

FINALLY:

if(rwlock != NULL){

pthread_rwlock_destroy(rwlock);

free(rwlock);

}

return result;

}

结果：
start2:1

start1:1

end2:1

end1:1

start2:2

start1:2

end2:2

end1:2

start2:3

start1:3

end2:3

end1:3

start2:4

start1:4

end2:4

end1:4

start2:5

start1:5

end2:5

end1:5

一次循环中start和end中的值都是一样的。

如果我们修改下程序：

[cpp] view
plain copy







#include <stdio.h>

#include <pthread.h>

#include <malloc.h>

#include <string.h>

int gi=0;

void* th_func1(void* arg){

int i;

for(i=0; i<5; i++){

printf("start1:%d\n",gi);

sleep(1);

printf("end1:%d\n",gi);

sleep(1);

}

}

void* th_func2(void* arg){

int i;

for(i=0; i<5; i++){

printf("start2:%d\n",gi);

sleep(1);

printf("end2:%d\n",gi);

sleep(1);

}

}

void* th_func3(void* arg){

int i;

for(i=0; i<5; i++){

gi++;

sleep(1);

}

}

int main(void){

int ret,result = 0;

pthread_t tid1,tid2,tid3;

ret = pthread_create(&tid1, NULL, th_func1, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid1);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_create(&tid2, NULL, th_func2, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid2);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_create(&tid3, NULL, th_func3, NULL);

if(ret != 0){

printf("pthread_create:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

ret = pthread_detach(tid3);

if(ret != 0){

printf("pthread_detach:%s\n",strerror(ret));

result = -1;

goto FINALLY;

}

sleep(15);

FINALLY:

return result;

}

运行结果：
start2:1

start1:1

end2:2

end1:2

start2:3

start1:3

end2:4

end1:4

start2:5

start1:5

end2:5

end1:5

start2:5

start1:5

end2:5

end1:5

start2:5

start1:5

end2:5

end1:5

一次循环中start和end中的值可能会不一样。

4.条件变量

条件变量是线程可用的另一种同步机制。条件变量给多个线程提供了一个回合的场所。条件变量与互斥量一起使用时，允许线程
以无竞争的方式等待特定的条件发生。
条件本身是由互斥量保护的，线程在改变条件状态前必须首先锁住互斥量，其他线程在获得互斥量之前不会觉察到这种改变，因为
必须锁住互斥量后才能计算条件。
条件变量使用之前必须先进行初始化，pthread_cond_t数据结构代表的条件变量可以用两种方式进行初始化，可以把常量
PTHREAD_COND_INITIALIZER赋给静态分配的条件变量，但是如果条件变量是动态分配的，可以使用pthread_cond_init函数初始化。
在释放底层的内存空间之前，可以使用pthread_cond_destroy函数对条件变量进行去除初始化。

[cpp] view
plain copy







#include<pthread.h>

int pthread_cond_init(pthread_cond_t *restrict cond, pthread_condattr_t *restrict attr);

int pthread_cond_destroy(pthread_cond_t *cond);

//成功则返回0，否则返回错误编号。

使用pthread_cond_wait等待条件变成真。

[cpp] view
plain copy







#include<pthread.h>

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);

//成功返回0，否则返回错误编号。

传递给pthread_cond_wait的互斥量对条件进行保护，调用者把锁住的互斥量传给函数。函数把调用线程放到等待条件的线程列表上，
然后对互斥量解锁，这两个操作时原子操作，这样就关闭了条件检查和线程进入休眠状态等待条件改变这两个操作之间的时间通道，
这样线程就不会错过条件的任何变化，pthread_cond_wait返回时，互斥量再次被锁住。

[cpp] view
plain copy







#include<pthread.h>

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex,

const struct timespec *restrict timeout);

//成功返回0，否则返回错误编号。

pthread_cond_timedwait函数的工作方式与pthread_cond_wait函数相似，只是多了一个timeout，timeout值指定了等待的时间。它
通过timespec结构指定。

[cpp] view
plain copy







struct timespec{

time_t tv_sec; //秒数

long tv_nsec; //纳秒

}

使用这个结构体时，需要指定等待多长时间，时间值是一个绝对值而不是相对值，例如如果要等待3分钟，就需要把当前时间加上3分钟
再转换到timespec结构。
如果时间到了但是条件还是没有出现，pthread_cond_timedwait将重新获取互斥量然后返回错误ETIMEDOUT。
从pthread_cond_wait或者pthread_cond_timedwait调用成功返回时，线程需要重新计算条件，因为其他的线程可能已经在运行并改变
了条件。

有两个函数可以用于通知线程条件已经满足。pthread_cond_signal函数将唤醒等待该线程的某个线程，而pthread_cond_broadcast函数
将唤醒等待该条件的所有线程。

[cpp] view
plain copy







#include <pthread.h>

int pthread_cond_siganl(pthread_cond_t *cond);

int pthread_cond_broadcast(pthread_cond_t *cond);

//成功则返回0，否则返回错误编号。

实践：

[cpp] view
plain copy







#include <stdio.h>

#include <pthread.h>

#include <string.h>

pthread_cond_t qready = PTHREAD_COND_INITIALIZER;

pthread_mutex_t qlock = PTHREAD_MUTEX_INITIALIZER;

int gi=0;

void* substract(void* arg){

int i;

for(i=0; i<5; i++){

pthread_mutex_lock(&qlock);

while(gi == 0)

pthread_cond_wait(&qready, &qlock);

printf("before substract gi:%d.\n",gi);

gi--;

printf("after substract gi:%d.\n",gi);

pthread_mutex_unlock(&qlock);

}

}

void* add(void* arg){

int i;

for(i=0; i<5; i++){

pthread_mutex_lock(&qlock);

printf("before add gi:%d.\n",gi);

gi++;

printf("after add gi:%d.\n",gi);

pthread_mutex_unlock(&qlock);

pthread_cond_signal(&qready);

}

}

int main(void){

int ret;

pthread_t tid1,tid2;

ret = pthread_create(&tid1, NULL, substract, NULL);

if(ret != 0){

printf("pthread_create:%s",strerror(ret));

return -1;

}

<span style="white-space:pre"> </span>sleep(1);

ret = pthread_create(&tid2, NULL, add, NULL);

if(ret != 0){

printf("pthread_create:%s",strerror(ret));

return -1;

}

ret = pthread_join(tid1,NULL);

if(ret != 0){

printf("pthread_join:%s",strerror(ret));

return -1;

}

ret = pthread_join(tid2,NULL);

if(ret != 0){

printf("pthread_join:%s",strerror(ret));

return -1;

}

return 0;

}

运行结果：
before add gi:0.

after add gi:1.

before substract gi:1.

after substract gi:0.

before add gi:0.

after add gi:1.

before substract gi:1.

after substract gi:0.

before add gi:0.

after add gi:1.

before substract gi:1.

after substract gi:0.

before add gi:0.

after add gi:1.

before substract gi:1.

after substract gi:0.

before add gi:0.

after add gi:1.

before substract gi:1.

after substract gi:0.

自旋锁

　　自旋锁与互斥量类似，但它不是通过休眠使进程阻塞，而是在获取锁之前一直处于忙等（自旋）状态。自旋锁可用于以下情况：锁被持有的时间短，而且线程并不希望在重新调度上花费太多的成本。

　　自旋锁通常作为底层原语用于实现其他类型的锁。根据它们所基于的系统体系结构，可以通过使用测试并设置指令有效地实现。当然这里说的有效也还是会导致CPU资源的浪费：当线程自旋等待锁变为可用时，CPU不能做其他的事情。这也是自旋锁只能够被位置一小段时间的原因。

　　自旋锁用在非抢占式内核中时是非常有用的：除了提供互斥机制以外，它们会阻塞中断，这样中断处理程序就不会让系统陷入死锁状态，因为它需要获取已被加锁的自旋锁（把中断想成另一种抢占）。在这种类型的内核中，中断处理程序不能休眠，因为它们能用的同步原语只能是自旋锁。

屏障

　　屏障（barrier）是用户协调多个线程并行工作的同步机制。屏障允许每个线程等待，直到所有的合作线程都到达某一点，然后从该点继续执行。我们已经看到一种屏障，pthread_join函数是一种屏障，允许一个线程等待，直到另一个线程退出。

　　但是屏障对象的概念更广，它们允许任意数量的线程等待，直到所有的线程处理完工作，而线程不需要退出。所有线程到达屏障后可以直接工作。

互斥锁和自旋锁的区别：

　　自旋锁是一种非阻塞锁，也就是说，如果某线程需要获取自旋锁，但该锁已经被其他线程占用时，该线程不会被挂起，而是在不断的消耗CPU的时间，不停的试图获取自旋锁。

　　互斥量是阻塞锁，当某线程无法获取互斥量时，该线程会被直接挂起，该线程不再消耗CPU时间，当其他线程释放互斥量后，操作系统会激活那个被挂起的线程，让其投入运行。

　　两种锁适用于不同场景：

　　如果是多核处理器，如果预计线程等待锁的时间很短，短到比线程两次上下文切换时间要少的情况下，使用自旋锁是划算的。

　　如果是多核处理器，如果预计线程等待锁的时间较长，至少比两次线程上下文切换的时间要长，建议使用互斥量。

　　如果是单核处理器，一般建议不要使用自旋锁。因为，在同一时间只有一个线程是处在运行状态，那如果运行线程发现无法获取锁，只能等待解锁，但因为自 身不挂起，所以那个获取到锁的线程没有办法进入运行状态，只能等到运行线程把操作系统分给它的时间片用完，才能有机会被调度。这种情况下使用自旋锁的代价 很高。

　　如果加锁的代码经常被调用，但竞争情况很少发生时，应该优先考虑使用自旋锁，自旋锁的开销比较小，互斥量的开销较大。