III.Linux系统编程_35线程_3线程间同步

1、第35章线程3.线程间同步3.1. mutex 多个线程同时访问共享数据时可能会冲突，这跟前面讲信号时所说的可重入性是 同样的问题。比如两个线程都要把某个全局变量增加 1,这个操作在某平台需要 三条指令完成：1. 从内存读变量值到寄存器2. 寄存器的值加13将寄存器的值写回内存假设两个线程在多处理器平台上同时执行这三条指令，则可能导致下图所示的结 果，最后变量只加了一次而非两次。图35.1.并行访问冲突銭程A的指令mov 0x8049540. %eax (eax = 5)add $0xlr %eax(eax = 6)mow %eaxr 0x8049540 (eax - 6)其它指令mov 0x

2、8049540,(eax = 5)add $0x1, %eax(eax = 6)mov %eax, DxS04Q540teax = 6)娈量i的内存单元的值5566思考一下，如果这两个线程在单处理器平台上执行，能够避免这样的问题吗?我们通过一个简单的程序观察这一现象。 上图所描述的现象从理论上是存在这种 可能的，但实际运行程序时很难观察到， 为了使现象更容易观察到，我们把上述 三条指令做的事情用更多条指令来做：val = coun ter;Iiprin tf(%x: %dn, (un sig ned| int)pthread_self(), val + 1);coun ter = val +

3、1;我们在“读取变量的值”和“把变量的新值保存回去”这两步操作之间插入一个 printf调用，它会执行write系统调用进内核，为内核调度别的线程执行提供 了一个很好的时机。我们在一个循环中重复上述操作几千次，就会观察到访问冲突的现象。! #inelude 11L#in clude #include r#defi ne NLOOP 5000! int coun ter;/*in creme ntedby| threads */1bL1Ik1void *doit(void *);rint main (i nt argc, char *argv)1 ipthread t tidA, tidB;l_

4、i1pthread create(&tidA, NULL, &doit,| NULL);pthread create(&tidB, NULL, &doit,| NULL);1/*wait for both threads to terminate*/pthreadoin( tidA, NULL);pthreadoi n(tidB, NULL);!111 b1kjreturn 0;j void *doit(void *vptr)!i 1int i, val;Ii1 i i/* Each thread fetches, prin ts, andin creme nts the coun ter N

5、LOOP times.* The value of the coun ter shouldin crease monotoni cally.*/for (i = 0; i 1 int pthread_mutex_lock(pthread_mutex_tL? *mutex);| int pthread_mutex_trylock(pthread_mutex_t j *mutex);i int pthread_mutex_unlock(pthread_mutex_t| *mutex);返回值：成功返回0，失败返回错误号。一个线程可以调用pthread_mutex_lock获得Mutex，如果这时另

6、一个线程已经 调用pthread_mutex_lock获得了该Mutex，则当前线程需要挂起等待，直到另 一个线程调用pthread_mutex_unlock释放Mutex，当前线程被唤醒，才能获得 该Mutex并继续执行。如果一个线程既想获得锁，又不想挂起等待，可以调用pthread_mutex_trylock ， 如果Mutex已经被另一个线程获得，这个函数会失败返回EBUSY，而不会使线 程挂起等待。现在我们用Mutex解决先前的问题:I #inelude I#in clude #in clude #define NLOOP 5000int coun ter;/*in creme nte

7、dbythreads */pthread mutex t coun ter mutex =PTHREAD_MUTEX_INITIALIZER;void *doit(void *);int main (i nt argc, char *argv)pthread_t tidA, tidB;pthread create(&tidA, NULL, doit,NULL);pthread create(&tidB, NULL, doit,NULL);/*wait for both threads to terminate*/pthread _join (tidA, NULL);pthreadoi n(ti

8、dB, NULL);r1return 0; i kiiIvoid *doit(void *vptr)! !int i, val;ii/*1* Each thread fetches, prin ts, andi in creme nts the coun ter NLOOP times. 1 i* The value of the coun ter should1 in crease monotoni cally.1k*/for (i = 0; i 0成立，然后其中一个线程置 mutex=O，而另一个线程并不知道这一情况，也置 mutex=O，于是两个线程都以为自己获得了锁。为了实现互斥锁操作

9、，大多数体系结构都提供了swap或exchange指令，该指令的作用是把寄存器和内存单元的数据相交换，由于只有一条指令，保证了原子性，即使是多处理器平台，访问内存的总线周期也有先后，一个处理器上的交换 指令执行时另一个处理器的交换指令只能等待总线周期。现在我们把lock和unlock的伪代码改一下（以x86的xchg指令为例）：! lock:Ij movb $0, %alIi xchgb %al, mutexif（al 寄存器的内容 0）return 0; else挂起等待;goto lock;i unlock:imovb $1, mutex唤醒等待Mutex的线程;return 0;unlo

10、ck中的释放锁操作同样只用一条指令实现，以保证它的原子性。也许还有读者好奇，“挂起等待”和“唤醒等待线程”的操作如何实现？每个 Mutex有一个等待队列，一个线程要在 Mutex上挂起等待，首先在把自己加入 等待队列中，然后置线程状态为睡眠，然后调用调度器函数切换到别的线程。- 个线程要唤醒等待队列中的其它线程， 只需从等待队列中取出一项，把它的状态 从睡眠改为就绪，加入就绪队列，那么下次调度器函数执行时就有可能切换到被 唤醒的线程。一般情况下，如果同一个线程先后两次调用lock，在第二次调用时，由于锁已经 被占用，该线程会挂起等待别的线程释放锁， 然而锁正是被自己占用着的，该线 程又被挂起而

11、没有机会释放锁，因此就永远处于挂起等待状态了，这叫做死锁(Deadlock )。另一种典型的死锁情形是这样：线程A获得了锁1,线程B获得了锁2,这时线程A调用lock试图获得锁2，结果是需要挂起等待线程 B释 放锁2，而这时线程B也调用lock试图获得锁1，结果是需要挂起等待线程 A 释放锁1,于是线程A和B都永远处于挂起状态了。不难想象，如果涉及到更 多的线程和更多的锁，有没有可能死锁的问题将会变得复杂和难以判断。写程序时应该尽量避免同时获得多个锁，如果一定有必要这么做，则有一个原则： 如果所有线程在需要多个锁时都按相同的先后顺序(常见的是按Mutex变量的地址顺序)获得锁，则不会出现死锁。

12、比如一个程序中用到锁1、锁2、锁3，它们所对应的Mutex变量的地址是锁1锁2锁3，那么所有线程在需要同时获 得2个或3个锁时都应该按锁1、锁2、锁3的顺序获得。如果要为所有的锁确 定一个先后顺序比较困难，则应该尽量使用pthread_mutex_trylock调用代替pthread_mutex_lock 调用，以免死锁。3.2. Con diti on Variable线程间的同步还有这样一种情况：线程 A需要等某个条件成立才能继续往下执 行，现在这个条件不成立，线程 A就阻塞等待，而线程B在执行过程中使这个 条件成立了，就唤醒线程A继续执行。在pthread库中通过条件变量(Conditi

13、on Variable )来阻塞等待一个条件，或者唤醒等待这个条件的线程。ConditionVariable用pthread_cond_t类型的变量表示，可以这样初始化和销毁：#in clude int pthread_cond_destroy(pthread_cond_t j *cond);iI int pthread_cond_init(pthread_cond_tI*restrict cond.const pthread_c on dattr_t *restrictattr);l! pthread_cond_t cond =! PTHREAD_COND_INITIALIZER;返回值：成

14、功返回0，失败返回错误号。和Mutex的初始化和销毁类似，pthread_condnit 函数初始化一个Condition Variable，attr参数为NULL则表示缺省属性，pthread_cond_destroy 函数销毁 一个Condition Variable。如果Condition Variable 是静态分配的，也可以用宏定义PTHEAD_COND_INITIALIZ初始化，相当于用pthread_cond_init 函数初始化 并且attr参数为NULL Condition Variable的操作可以用下列函数：#in clude int pthread_c on d_tim

15、edwait(pthread_c ond_t*restrict cond.pthread_mutex_t *restrict mutex,const struct timespec *restrictabstime);int pthread cond wait(pthread cond t*restrict cond,pthread_mutex_t *restrict mutex);1 int pthread c on d broadcast(pthread c ond t*cond);int pthread c on d sig nal(pthread c ond t! *cond);返回值

16、：成功返回0，失败返回错误号。可见，一个Condition Variable总是和一个Mutex搭配使用的。一个线程可以调 用pthread_cond_wait在一个Condition Variable上阻塞等待，这个函数做以下 三步操作：1. 释放 Mutex2. 阻塞等待3. 当被唤醒时，重新获得 Mutex并返回pthread_co nd_timedwait函数还有一个额外的参数可以设定等待超时，如果到 达了 abStime亦指定的时刻仍然没有别的线程来唤醒当前线程，就返回ETIMEDOUT一个线程可以调用 pthread_cond_signal 唤醒在某个 Condition Vari

17、able上等待的另一个线程，也可以调用pthread_cond_broadcast 唤醒在这个Condition Variable 上等待的所有线程。下面的程序演示了一个生产者-消费者的例子，生产者生产一个结构体串在链表 的表头上，消费者从表头取走结构体。#in clude #in elude vpthread.h1 #in elude !111 b1kj struct msg Istruct msg *n ext;int num;! ;!| struct msg *head; I ii pthread cond t has product =j PTHREAD COND INITIALIZE

18、R; i j1 pthread mutex t lock =I PTHREAD MUTEX INITIALIZER;i1k1kvoid *con sumer(void *p)f struct msg *mp;ifor (;) 1pthread_mutex_lock (&l ock);while (head = NULL)Lpthread c on d wait (&has product,i &lock);Imp = head;head = mp-n ext;pthread_mutex_ un lock(&lock);iprin tf(C on sume%dn,mp-nu m);free(mp

19、);sleep(ra nd() % 5);void *producer(void *p)struct msg *mp;for (;) mp = malloc(sizeof(structmsg);mp-n um = rand() % 1000 + 1;prin tf(Produce%dn,mp-nu m);pthread_mutex_lock (&l ock);mp-n ext = head;head = mp;pthread_mutex_ un lock(&lock);pthread_c on d_sig nal (&has_product);sleep(ra nd() % 5);int ma

20、in (i nt argc, char *argv)pthread_t pid, cid;sran d(time(NULL);| pthread_create(&pid, NULL, producer, j NULL);Ipthread_create(&cid, NULL, con sumer,j NULL);pthread _joi n( pid, NULL);pthread _join (cid, NULL);return 0;ll1I执行结果如下:$ ./a.outProduce 744Con sume 744Ij Produce 567h :Produce 8811 Con sume

21、881iiProduce 911! Consume 911s!Ii Consume 567b | Produce 698ii Consume 698习题1、在本节的例子中，生产者和消费者访问链表的顺序是LIFO的，请修改程序,把访问顺序改成FIFO。3.3. SemaphoreMutex变量是非0即1的，可看作一种资源的可用数量，初始化时 Mutex是1， 表示有一个可用资源，加锁时获得该资源，将Mutex减到0,表示不再有可用资 源，解锁时释放该资源，将 Mutex重新加到1，表示又有了一个可用资源。信号量(Semaphore )和Mutex类似，表示可用资源的数量，和 Mutex不同的 是

22、这个数量可以大于1。本节介绍的是POSIX semaphore 库函数，详见sem_overview(7)，这种信号量 不仅可用于同一进程的线程间同步，也可用于不同进程间的同步。#include iIj I int sem_init(sem_t *sem, int pshared,Ij unsigned int value);Hint sem_wait(sem_t *sem);ui int sem_trywait(sem_t *sem);Iint sem_post(sem_t * sem);IIint sem_destroy(sem_t * sem);semaphore 变量的类型为 sem_

23、t ,sem_init()初始化一个 semaphore 变量，value 参数表示可用资源的数量，pshOred参数为0表示信号量用于同一进程的线程间 同步，本节只介绍这种情况。在用完semaphore变量之后应该调用sem_destroy() 释放与semaphore相关的资源。调用sem_wait()可以获得资源，使semaphore的值减1，如果调用sem_wait() 时semaphore的值已经是0，则挂起等待。如果不希望挂起等待，可以调用 sem_trywait()。调用sem_post()可以释放资源，使 semaphore的值加1，同时 唤醒挂起等待的线程。一 上一节生产者-

24、消费者的例子是基于链表的，其空间可以动态分配，现在基于固 定大小的环形队列重写这个程序：#in clude #in elude vpthread.h#in elude #in elude #defi ne NUM 5int queueNUM;sem_t bla nk_nu mber, product nu mber;void *producer(void *arg)int p = 0;while (1) sem_wait (&bla nk_nu mber);queuep = ran d() % 1000 + 1;printf(Produce %dn,queuep);sem_post (&productnu mber);p = (p+1)%NUM;sleep(ra nd()%5);void *con sumer(void *arg)int c = 0;while sem_wait (&product, nu mber);!;prin tf(Co nsume %dn.i queuec);1queuec = 0;11sem post (&bla nk nu mber);cc = (c+1)%NUM;sleep(ra nd()%5);i 1il