IPC 互斥锁和条件变量

在多线程或多进程间共享数据时，为了保证共享数据的正确性，同步是必不可少的。其中，互斥锁与条件变量是同步的重要组成部分, 互斥锁用来上锁， 条件变量用来等待。

1.1 互斥锁

互斥是指相互排斥(mutual exclusion)。互斥锁用于保护临界区(critical region), 以保证任何时刻只有一个线程（进程）在执行其中的代码，假设互斥锁由多个线程（进程），保护一个临界区的代码通常如下:

lock_the_mutex(...);
/* critical region */
unlock_the_mutex(...);

1.1.1 互斥锁初始化

Posix 互斥锁被声明为 pthread_mutex_t 数据类型的变量。可以用如下两种方法进行初始化。

• 将互斥量初始化成常量值 PTHREAD_MUTEX_INITIALIZER 。linux 中 PTHREAD_MUTEX_INITIALIZER 的宏定义如下：

  #if __WORDSIZE == 64
  #define PTHREAD_MUTEX_INITIALIZER               \
      { { 0, 0, 0, 0, 0, 0, { 0, 0 } } }
  #ifdef __USE_GNU
  #define PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP                  \
      { { 0, 0, 0, 0, PTHREAD_MUTEX_RECURSIVE_NP, 0, { 0, 0 } } }
  #define PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP                         \
      { { 0, 0, 0, 0, PTHREAD_MUTEX_ERRORCHECK_NP, 0, { 0, 0 } } }
  #define PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP                   \
      { { 0, 0, 0, 0, PTHREAD_MUTEX_ADAPTIVE_NP, 0, { 0, 0 } } }
  #endif
  #else
  #define PTHREAD_MUTEX_INITIALIZER               \
      { { 0, 0, 0, 0, 0, { 0 } } }
  #ifdef __USE_GNU
  #define PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP                  \
      { { 0, 0, 0, PTHREAD_MUTEX_RECURSIVE_NP, 0, { 0 } } }
  #define PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP                 \
      { { 0, 0, 0, PTHREAD_MUTEX_ERRORCHECK_NP, 0, { 0 } } }
  #define PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP                   \
      { { 0, 0, 0, PTHREAD_MUTEX_ADAPTIVE_NP, 0, { 0 } } }
  #endif
  #endif
  

• 如果互斥锁是通过动态分配的，那么我们在运行之时就不用使用第一种方式来初始化互斥锁，这时可以通过 pthread_mutex_init 函数来初始化。同时从上面的 PTHREAD_MUTEX_INITIALIZER 的宏定义可以看出，我们只要 将动态分配的变量全都初始化为0，也是行的，linux下可以用 bzero，标准 C 中有 memset。

1.1.2 互斥锁上锁和解锁

三个上锁或解锁的函数:

#include <pthread.h>

int pthread_mutex_lock(pthread_mutex_t *mptr);

int pthread_mutex_trylock(pthread_mutex_t *mptr);

int pthread_mutex_unlock(pthread_mutex_t *mptr);

其中pthread_mutex_lock为阻塞函数，pthread_mutex_trylock 为非阻塞函数。对阻塞函数而言，如果互斥锁己被上锁则返回EBUSY错误。

 

1.2 条件变量

1.2.1 等待与信号发送

条件变量是类型为 pthread_cond_t 的变量。条件变量的相关操作函数如下:

#include<pthread.h>

int pthread_cond_wait(pthread_cond_t *cptr, pthread_mutex_t *mptr);
int pthread_cond_signal(pthread_cond_t *cptr);

pthread_cond_wait(cond, mutex) 该函数原子地执行以下两个动作：

• 给互斥锁 mutex 解锁。
• 把调用线程投入睡眠， 直到另外某个线程就本条件变量cond调用pthread_cond_signal，pthread_cond_wait 在返回前重新给互斥锁 mutex 上锁。

给条件变量发送信号的代码大体如下：

struct {
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    /* 维护本条件的各个变量 */
}var = { PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER /*, ... */};

pthread_mutex_lock(& var.mutex);
/* set the condition as true */
pthread_cond_signal(& var.cond);
pthread_mutex_unlock(& var.mutex);
  

为了避免虚假唤醒(spurious wakeup)， 测试条件并进入睡眠以等待该条件变为真 的代码如下：

pthread_mutex_lock(& var.mutex);
while (/* condition is false */ )
    pthread_cond_wait(& var.cond, & var.mutex);
/* change the condition */
pthread_mutex_unlock(& var.mutex);

为了避免上锁冲突将

/* producer */
pthread_mutex_lock(& nready.mutex);
if( nready.nready == 0)
    pthread_cond_signal(& nready.cond);
nready.nready++;
pthread_mutex_unlock(& nready.mutex);

/* consumer */
pthread_mutex_lock(& nready.mutex);
while( nready.nready == 0)
    pthread_cond_wait(& nready.cond, & nready.mutex);
nready.nready--;
pthread_mutex_unlock(& nready.mutex);

中的生产者的代码改为：

int dosignal;

pthread_mutex_lock(& nready.mutex);
dosignal = ( nready.nready == 0);
nready.nready++;
pthread_mutex_unlock(& nready.mutex);
if (dosignal)
    pthread_cond_signal(& nready.cond);

1.2.2 定时等待与广播

pthread_cond_signal 只唤醒等待在相应条件变量的一个线程，pthread_cond_broadcast 可以唤醒阻塞在相应条件变量上的所有线程。

#include <pthread.h>
int pthread_cond_broadcast(pthread_cond_t *cptr);
int pthread_cond_timedwait(pthread_cont_t *cptr, pthread_mutex_t *mptr, \
                           const struct timespec *abstime);

struct timespec{
    time_t tv_sec; /* seconds */
    long tv_nsec;  /* nanoseconds */
};

1.3 互斥锁和条件变量的属性

 

互斥锁属性的数据类型为 pthread_mutexattr_t, 条件变量属性的数据类型为 pthread_condattr_t。

互斥锁和条件变量及其性性初始化、摧毁函数和属性设置函数如下：

int pthread_mutex_init(pthread_mutex_t *mptr, const pthread_mutexattr_t *attr);
int pthread_mutex_destroy(pthread_mutex_t *mptr);
int pthread_cond_init(pthread_cond_t *cptr, const pthread_condattr_t *attr);
int pthread_cond_destroy(pthread_cond_t *cptr);

int pthread_mutexattr_init(pthread_mutexattr_t *mptr);
int pthread_mutexattr_destroy(pthread_mutexattr_t *mptr);
int pthread_condattr_init(pthread_condattr_t *cptr);
int pthread_condattr_destroy(pthread_condattr_t *cptr);

int pthread_mutexattr_getpshared(const pthread_mutexattr_t *attr, int *valptr);
int pthread_mutexattr_setpshared(pthread_mutexattr_t *attr, int value);
int pthread_condattr_getpshared(const pthread_condattr_t *attr, int *valptr);
int pthread_condattr_setpshared(pthread_condattr_t *attr, int value);

/* return 0 if successed, else return errno */