你的位置:首页 > 操作系统

[操作系统]APUE学习之多线程编程(二):线程同步

     为了保证临界资源的安全性和可靠性,线程不得不使用锁,同一时间只允许一个或几个线程访问变量。常用的锁有互斥量,读写锁,条件变量
    
     一、互斥量
     互斥量是用pthread_mutex_t数据类型表示的,在使用之前,必须对其进行初始化,可以把它设置为PTHREAD_MUTEX_INITIALIZER(只适于静态分配的互斥量),也可以通过调用pthread_mutex_init函数进行初始化,最后还要调用pthread_mutex_destroy进行释放。
#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);
     要用默认的属性初始化互斥量,只需把attr设为NULL,后面在讨论互斥量属性。
     对互斥量进行加锁,使用pthread_mutex_lock,如果互斥量已经上锁,调用线程将阻塞至互斥量解锁,对互斥量解锁,使用pthread_mutex_unlock,如果线程不希望被阻塞,它可以调用pthread_mutex_trylock尝试对互斥量进行加锁,如果互斥量未锁住,则成功加锁,如果互斥量已锁住,pthread_mutex_trylock就会失败,返回EBUSY。
#include <pthread.h>int pthread_mutex_lock(pthread_mutex_t *mutex);int pthread_mutex_unlock(pthread_mutex_t *mutex);int pthread_mutex_trylock(pthread_mutex_t *mutex);
例子:
#include <stdio.h>#include <pthread.h> struct foo{  int f_count;  pthread_mutex_t f_lock;  int f_id;}; struct foo * foo_alloc(int id){  struct foo *fp = NULL;   if ((fp = malloc(sizeof(struct foo))) != NULL)  {    fp->f_count = 1;    fp->f_id = id;    if (pthread_mutex_init(&fp->f_lock, NULL) != 0)    {      free(fp);      return NULL;    }  }   return fp;} void foo_hold(struct foo *fp){  pthread_mutex_lock(&fp->f_lock);  fp->f_count++;  pthread_mutex_unlock(&fp->f_lock);} void foo_rele(struct foo *fp){  pthread_mutex_lock(&fp->f_lock);   if (--fp->f_count == 0)  {    pthread_mutex_unlock(&fp->f_lock);    pthread_mutex_destroy(&fp->f_lock);    free(fp);  }  else  {    pthread_mutex_unlock(&fp->f_lock);  }}

View Code     上面的例子描述了用于保护某个数据结构的互斥量,我们在对象中嵌入引用计数,确保在所有使用该对象的线程完成数据访问之前,该对象的内存空间不会被释放。
 
     如果线程对同一个互斥量加锁两次,那么它自身将陷入死锁状态。如果有一个以上的互斥量,且允许一个线程一直占有第一个互斥量,并且试图锁住第二个互斥量时处于阻塞状态,但是拥有第二个互斥量的线程也在试图锁住第一个互斥量,也阻塞,就死锁了。
     可以通过仔细控制互斥量加锁的顺序来避免死锁的发生,譬如要求所有线程必须先锁住互斥量A才能锁住互斥量B。另一种办法是当线程无法获得下一个互斥量的时候,就释放自己已占有的互斥量,过一段时间再试。
例子:
#include "apue.h"#include <pthread.h> #define NMASH 29#define HASH(id) (((unsigned long)id) % NMASH) struct foo *fh[NMASH]; pthread_mutex_t hashlock = PTHREAD_MUTEX_INITIALIZER; struct foo{  int f_count;  pthread_mutex_t f_lock;  int f_id;  struct foo *f_next;}; struct foo *foo_alloc(int id){  struct foo *fp = NULL;  int idx = 0;   if ((fp = malloc(sizeof(struct foo))) != NULL)  {    fp->f_count = 1;    fp->f_id = if;    if (pthread_mutex_init(&fp->f_lock, NULL) != 0)    {      free(fp);      return NULL;    }     idx = HASH(id);    pthread_mutex_lock(&hashlock);    fp->f_next = fh[idx];    fh[idx] = fp;    pthread_mutex_lock(&fp->f_lock);    pthread_mutex_unlock(&hashlock);    pthread_mutex_unlock(&fp->f_lock);  }   return fp;} void foo_hold(struct foo *fp){  pthread_mutex_lock(&fp->f_lock);  fp->f_count++;  pthread_mutex_unlock(&fp->f_lock);} struct foo *foo_find(int id){  struct foo *fp = NULL;   pthread_mutex_lock(&hashlock);   for (fp = fh[HASH(id)]; fp != NULL; fp = fp->next)  {    if (fp->f_id = id)    {      foo_hold(fp);      break;    }  }   pthread_mutex_unlock(&hashlock);  return fp;} void foo_rele(struct foo *fp){  struct foo *tfp = NULL;  int idx = 0;   pthread_mutex_lock(&fp->f_lock);   if (fp->f_count == 1)  {    pthread_mutex_unlock(&fp->f_lock);    pthread_mutex_lock(&hashlock);    pthread_mutex_lock(&fp->f_lock);     if (fp->f_count != 1)    {      fp->f_count--;      pthread_mutex_unlock(&hashlock);      pthread_mutex_unlock(&fp->f_lock);      return;    }     idx = HASH(fp->f_id);    tfp = fh[idx];    if (tfp = fp)    {      fh[idx] = fp->f_next    }    else    {      while(tfp->next != fp)      {        tfp = tfp->next;      }      tfp->next = fp->f_next;    }     pthread_mutex_unlock(&hashlock);    pthread_mutex_unlock(&fp->f_lock);    pthread_mutex_destroy(&fp->f_lock);    free(fp);  }  else  {    fp->f_count--;    pthread_mutex_unlock(&fp->f_lock);  }}

View Code     这个例子比上一个例子多了一个散列表和一个保护散列表的互斥量,加锁的顺序是先hashlock,再f_lock,注意这个顺序,就不会发生死锁,不过这样也导致代码太繁琐,最后一个函数解锁f_lock后重新加锁f_lock,需要重新考察f_count的值,因为可能在这期间被其他线程修改。
     这样的方式太复杂,让hashlock也保护f_cout,事情会简单很多。
例子:
#include "apue.h"#include <pthread.h> #define NMASH 29#define HASH(id) (((unsigned long)id) % NMASH) struct foo *fh[NMASH]; pthread_mutex_t hashlock = PTHREAD_MUTEX_INITIALIZER; struct foo{  int f_count;  pthread_mutex_t f_lock;  int f_id;  struct foo *f_next;}; struct foo *foo_alloc(int id){  struct foo *fp = NULL;  int idx = 0;   if ((fp = malloc(sizeof(struct foo))) != NULL)  {    fp->f_count = 1;    fp->f_id = if;    if (pthread_mutex_init(&fp->f_lock, NULL) != 0)    {      free(fp);      return NULL;    }     idx = HASH(id);    pthread_mutex_lock(&hashlock);    fp->f_next = fh[idx];    fh[idx] = fp;    pthread_mutex_lock(&fp->f_lock);    pthread_mutex_unlock(&hashlock);    pthread_mutex_unlock(&fp->f_lock);  }   return fp;} void foo_hold(struct foo *fp){  pthread_mutex_lock(&hashlock);  fp->f_count++;  pthread_mutex_unlock(&hashlock);} struct foo *foo_find(int id){  struct foo *fp = NULL;   pthread_mutex_lock(&hashlock);   for (fp = fh[HASH(id)]; fp != NULL; fp = fp->next)  {    if (fp->f_id = id)    {      foo_hold(fp);      break;    }  }   pthread_mutex_unlock(&hashlock);  return fp;} void foo_rele(struct foo *fp){  struct foo *tfp = NULL;  int idx = 0;   pthread_mutex_lock(&hashlock);   if (fp->f_count == 1)  {     idx = HASH(fp->f_id);    tfp = fh[idx];    if (tfp = fp)    {      fh[idx] = fp->f_next    }    else    {      while(tfp->next != fp)      {        tfp = tfp->next;      }      tfp->next = fp->f_next;    }     pthread_mutex_unlock(&hashlock);    pthread_mutex_destroy(&fp->f_lock);    free(fp);  }  else  {    fp->f_count--;    pthread_mutex_unlock(&hashlock);  }}

View Code     当线程试图获取一个已加锁的互斥量时,pthread_mutex_timedlock互斥量原语允许绑定线程阻塞时间。pthread_mutex_timedlock和pthread_mutex_lock是基本等价的,但是达到超时时间后,pthread_mutex_timedlock会返回。超时时间指原意等待的绝对时间。这个超时时间是用timespec来表示的
#include <pthread.h>#include <time.h>int pthread_mutex_timedlock(pthread_mutex_t *restrict mutex, const struct timespec *restrict tsptr);

 


     二、读写锁
     读写锁与互斥量相似,不过读写锁允许更高的并行性,一次只有一个线程可以占有写模式的读写锁,但是多个线程可以同时占有读模式的读写锁,简单地来说,就说支持一个写者,多个读者。
     当读写锁是写加锁状态时,所以试图对这个锁加锁的线程都会被阻塞,当读写锁在读加锁状态时,所以试图以读模式对它进行加锁的线程都可以得到访问权,但是希望以写模式加锁的线程会被阻塞。不过当有一个线程企图以写模式获取锁时,读写锁会阻塞后面的读模式锁请求,防止读模式锁长期占用。
     可知,读写锁适用于对数据结构读的次数远大于写的情况,又称共享互斥锁,读共享,写互斥。
#include <pthread.h>int pthread_rwlock_init(pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
     读写锁调用phtread_rwlock_init进行初始化,如果希望读写锁有默认的属性,传null给attr即可。
     读的模式下锁定读写锁,需要调用phtread_rwlock_rdlock,写的模式下锁定读写锁,需要调用pthread_rwlock_wrlock,不过以何种方式锁定读写锁,都可以调用pthread_rwlock_unlock解锁。
#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);
例子:
#include <stdio.h>#include <pthread.h> struct job{  struct job *j_next;  struct job *j_prev;  pthread_t j_id;}; struct queue{  struct job *q_head;  struct job *q_tail;  pthread_rwlock_t q_lock;}; int queue_init(struct queue *qp){  int err;   qp->q_head = NULL;  qp->q_tail = NULL;  err = pthread_rwlock_init(&qb->q_lock, NULL);  if (err != 0)  {    return err;  }   return 0} void job_insert(struct queue *qp, struct job *jp){  pthread_rwlock_wrlock(&qb->q_lock);  jp->next = qp->head;  jp->j_prev = NULL;   if (qp->q_head != NULL)  {    qp->q_head->j_prev = jp;  }  else  {    qp->tail = jp;  }  qp->head = jp;  pthread_rwlock_unlock(&qp->q_lock);} void job_append(struct queue *qp, struct job *jp){  pthread_rwlock_wrlock(&qp->q_lock);  jp->j_next = NULL;  jp->j_prev = qp->tail;  if (qp->q_tail != NULL)  {    qp->q_tail->j_next = jp;  }  qp->q_tail = jp;  pthread_rwlock_unlock(&qp->q_lock);} void job_remove(struct queue *qp, struct job *jp){  pthread_rwlock_wrlock(&qp->q_lock);  if (jp == qp->q_head)  {    qp->q_head = jp->j_next;    if (qp->q_tail == jp)    {      qp->tail = NULL;    }    else    {      jp->next->j_prev = jp->j_prev;    }  }  else if (jp == qp->q_tail)  {    qp->q_tail = jp->j_prev;    jp->j_prev->j_next = NULL;  }  else  {    jp->j_prev->j_next = jp->j_next;    jp->j_next->j_prev = jp->j_prev;  }  pthread_rwlock_unlock(&qp->q_lock);} struct job *job_find(struct queue *qp, pthread_t id){  struct job *jp;   if (pthread_rwlock_rdlock(&qp->q_lock) != 0)  {    return NULL;  }   for (jp = qb->q_head; jp != NULL; jp = jp->j_next)  {    if (pthread_equal(jp->j_id, id))    {      break;    }  }  pthread_rwlock_unlock(&qp->q_lock);  return jp;}

View Code     与互斥量一样,读写锁也有带超时的读写锁函数,避免陷入永久的阻塞。
#include <pthread.h>#include <time.h>int pthread_rwlock_timedrdlock(pthread_rwlock_t *restrict rwlock, const struct timespec *restrict tsptr);int pthread_rwlock_timedwrlock(pthread_rwlock_t *restrict rwlock, const struct timespec *restrict tsptr);

 


     三、条件变量
     条件变量与互斥量一起使用时,允许线程以无竞争的方式等待特定的条件发生。
     条件本身由互斥量保护,线程在改变条件状态之前必须锁定互斥量。在使用条件变量之前,必须把它初始化,可以把常量PTHREAD_CON_INITIALIZE赋给静态分配的条件变量,也可用pthread_cond_init函数进行初始化。使用pthread_cond_destroy释放。
#include <pthread.h>int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);int pthread_cond_destroy(pthread_con_t *cond);
     如果需要一个默认属性的条件变量,把null给attr即可。
     我们使用pthread_cond_wait等待条件变量为真,如果在给定时间内不能满足,则返回错误码。
#include<pthread.h>int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex)int pthread_cond_timedwait(pthread_cond_t *restrict cond, phtread_mutex_t *restrict mutex, const struct timespec *restrict tsptr)
     调用者把锁定的互斥量传给函数,函数自动把调用线程放到等待条件的线程列表上,对互斥量解锁,当pthread_cond_wait返回时,互斥量再次被锁住。pthread_cond_timedwait多了原意等待的时间。
     有两个函数可用于通知线程条件已满足,pthread_cond_signal函数至少唤醒一个,pthread_cond_broadcast唤醒等待该条件的所有线程。
#include<phtread.h>int pthread_cond_signal(pthread_cond_t *cond)int pthread_cond_broadcast(pthread_cond_t *cond)
例子:
#include <pthread.h> struct msg{  struct msg *m_next;}; struct msg *workq; pthread_cond_t qready = PTHREAD_COND_INITIALIZER;pthread_mutex_t qlock = PTHREAD_MUTEX_INITIALIZER; void process_msg(void){  struct msg *mp;   for(;;)  {    pthread_mutex_lock(&qlock);    while (workq == NULL)    {      pthread_cond_wait(&qready, &qlock);    }     mp = workq;    workq = mp->m_next;    pthread_mutex_unlock(&qlock);  }} void enqueue_msg(struct msg *mp){  pthread_mutex_lock(&qlock);  mp->m_next = workq;  workq = mp;  pthread_mutex_unlock(&qlock);  pthread_cond_signal(&qready);}

View Code