wangpifu

继上一篇文章学习了一些基础的api后，这一章我们接着来看pthread所提供的锁,条件变量

Mutex Variable(互斥锁)

Overview

Mutex是"mutual exclusion"(互斥)简称.Mutex variable 就像一把锁一样保护共享数据资.mutex的基本概念就是,在任何时候只有一个线程能 lock一个mutex 变量。所以，即使很多线程尝试去锁一个mutex，也仅仅只有一个线程能成功。

创建，销毁，锁定与释放

pthread_mutex_init(pthread_mutex_t * mutex, const phtread_mutexattr_t * mutexattr);//动态方式创建锁，相当于new动态创建一个对象
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;//以静态方式创建锁
pthread_mutex_destory(pthread_mutex_t *mutex)//释放互斥锁，相当于delete
pthread_mutex_lock(pthread_mutex_t *mutex)
//以阻塞方式运行的。如果之前mutex被加锁了，那么程序会阻塞在这里。
pthread_mutex_unlock(pthread_mutex_t *mutex)
int pthread_mutex_trylock(pthread_mutex_t * mutex);
// 会尝试对mutex加锁。如果mutex之前已经被锁定，返回非0,；如果mutex没有被锁定，则函数返回并锁定mutex.该函数是以非阻塞方式运行。也就是说如果mutex之前已经被锁定，函数会返回非0，程序继续往下执行。
典型使用mutex的顺序如下：

1. 创建和初始化 mutex 变量;
2. 许多线程尝试锁住 mutex;
3. 只有一个线程成功锁住 mutex,其他线程等待;
4. 拥有 mutex 的线程进行自己的操作;
5. 拥有线程解锁 mutex;
6. 其他线程继续获取 mutex 并持续如上步骤;
7. 最后 mutex 销毁.

互斥锁属性设置

int pthread_mutexattr_init(pthread_mutexattr_t *mattr);
//成功返回0，其它返回值表示出错  
int pthread_mutexattr_destroy(pthread_mutexattr_t *mattr);
//成功返回0，其它返回值表示出错

互斥锁具有一些属性，通过修改这些属性可以控制锁的一些行为。缺省的互斥锁属性及其值如下：

1. pshared: PTHREAD_PROCESS_PRIVATE
2. type: PTHREAD_MUTEX_DEFAULT
3. protocol: PTHREAD_PRIO_NONE
4. prioceiling: –
5. robustness: PTHREAD_MUTEX_STALLED_NP
   可以用pthread_mutexattr_init将与互斥锁对象相关联的属性初始化为其缺省值。pthread_mutexattr_init的参数类型实际上是opaque的，其中包含一个由系统分配的属性对象。该函数执行过程中会为属性对象分配所需的内存，因而如果未通过pthread_mutexattr_destroy销毁互斥锁属性对象时就会导致内存泄漏。
   对于互斥锁属性对象,必须首先通过调用pthread_mutexattr_destroy将其销毁,才能重新初始化该对象。

作用域

int pthread_mutexattr_setpshared(pthread_mutexattr_t *mattr,int pshared);
//成功返回0，其它返回值表示出错  
int pthread_mutexattr_getpshared(pthread_mutexattr_t *mattr,int *pshared);
//成功返回0，其它返回值表示出错

函数pthread_mutexattr_setpshared用来设置互斥锁的作用域。互斥锁变量可以是进程专用的变量,也可以是跨越进程边界的变量。
范围属性的取值及其含义：
PTHREAD_PROCESS_SHARED：具有该属性的互斥锁可以在多个进程中的线程之间共享。
PTHREAD_PROCESS_PRIVATE：只有创建本互斥锁的线程所在的进程内的线程才能够使用该互斥锁变量。该值是缺省值。
函数pthread_mutexattr_getpshared可用来返回由pthread_mutexattr_setpshared设置的互斥锁变量的范围。

类型属性

int pthread_mutexattr_settype(pthread_mutexattr_t *attr , int type);
//成功返回0，其它返回值表示出错  
int pthread_mutexattr_gettype(pthread_mutexattr_t *attr , int *type);
//成功返回0，其它返回值表示出错  

pthread_mutexattr_settype用来设置指定互斥锁的类型属性。类型属性的缺省值为PTHREAD_MUTEX_DEFAULT。
互斥锁的类型及其行为：

• PTHREAD_MUTEX_NORMAL：不提供死锁检测。如果一个线程试图对一个互斥锁重复锁定，将会引起这个线程的死锁。如果试图解锁一个由别的线程锁定的互斥锁会引发不可预料的结果。如果一个线程试图解锁已经被解锁的互斥锁也会引发不可预料的结果
• PTHREAD_MUTEX_ERRORCHECK：提供错误检查。如果一个线程试图对一个互斥锁重复锁定，将会返回一个错误代码。如果试图解锁一个由别的线程锁定的互斥锁将会返回一个错误代码。如果一个线程试图解锁已经被解锁的互斥锁也将会返回一个错误代码
• PTHREAD_MUTEX_RECURSIVE：如果一个线程对这种类型的互斥锁重复上锁，不会引起死锁，一个线程对这类互斥锁的多次重复上锁必须由这个线程来重复相同数量的解锁，这样才能解开这个互斥锁，别的线程才能得到这个互斥锁。如果试图解锁一个由别的线程锁定的互斥锁将会返回一个错误代码。如果一个线程试图解锁已经被解锁的互斥锁也将会返回一个错误代码。这种类型的互斥锁只能是进程私有的（作用域属性为PTHREAD_PROCESS_PRIVATE）
• PTHREAD_MUTEX_DEFAULT：这种类型的互斥锁不会自动检测死锁。如果一个线程试图对一个互斥锁重复锁定，将会引起不可预料的结果。如果试图解锁一个由别的线程锁定的互斥锁会引发不可预料的结果。如果一个线程试图解锁已经被解锁的互斥锁也会引发不可预料的结果。POSIX标准规定，对于某一具体的实现，可以把这种类型的互斥锁定义为其他类型的互斥锁
  在linux中互斥锁的相关类型定义如下（最好的办法是检查pthread.h这个头文件）：

  #if defined __USE_UNIX98 || defined __USE_XOPEN2K8
  PTHREAD_MUTEX_NORMAL = PTHREAD_MUTEX_TIMED_NP,
  PTHREAD_MUTEX_RECURSIVE = PTHREAD_MUTEX_RECURSIVE_NP,
  PTHREAD_MUTEX_ERRORCHECK = PTHREAD_MUTEX_ERRORCHECK_NP,
  PTHREAD_MUTEX_DEFAULT = PTHREAD_MUTEX_NORMAL
  #endif

pthread_mutexattr_gettype用来获取由pthread_mutexattr_settype设置的互斥锁的类型属性。

协议属性

int pthread_mutexattr_setprotocol(pthread_mutexattr_t *attr, int protocol);
//成功返回0，其它返回值表示出错  
int pthread_mutexattr_getprotocol(const pthread_mutexattr_t *attr, int *protocol); //成功返回0，其它返回值表示出错   

pthread_mutexattr_setprotocol用来设置互斥锁的协议属性。
互斥锁协议属性的可能值及其含义：

• PTHREAD_PRIO_NONE： 线程的优先级和调度不会受到互斥锁拥有权的影响.
• PTHREAD_PRIO_INHERIT： 此协议值会影响拥有该互斥锁的线程的优先级和调度。如果更高优先级的线程因thread1所拥有的一个或多个互斥锁而被阻塞,而这些互斥锁是用 PTHREAD_PRIO_INHERIT 初始化的,则thread1的运行优先级为优先级pri1和优先级pri2中优先级较高的那一个.(其中 thread1的优先级为pri1,所有正在等待这些互斥锁的线程的最高优先级为pri2.)如果thread1因另一个线程(thread3) 拥有的互斥锁而被阻塞,则相同的优先级继承效应会以递归方式传播给thrd3。
  使用PTHREAD_PRIO_INHERIT可以避免优先级逆转。当低优先级的线程持有较高优先级线程所需的锁时,就会发生优先级逆转。此时只有在较低优先级的线程释放该锁之后,较高优先级的线程才能继续执行。如果没有优先级继承，低优先级的线程可能会在很长一段时间内都得不到调度，而这会导致等待低优先级线程锁持有的锁的高优先级线程也等待很长时间（因为低优先级线程无法运行，因而就无法释放锁，所以高优先级线程只能继续阻塞在锁上）。使用优先级继承可以短时间的提高低优先级线程的优先级，从而使它可以尽快得到调度，然后释放锁。低优先级线程在释放锁后就会恢复自己的优先级。
• PTHREAD_PRIO_PROTECT：当线程拥有一个或多个使用PTHREAD_PRIO_PROTECT 初始化的互斥锁时,线程的优先级和调度会受到影响。线程将以优先级pri1和优先级pri2中优先级较高的那一个优先级来运行，其中该线程的优先级为pri1，所有该线程持有的锁的最高优先级为pri2，被该线程所持有的锁阻塞的高优先级线程对该线程的调度没有影响。

PTHREAD_PRIO_INHERIT 和 PTHREAD_PRIO_PROTECT 只有在采用实时调度策略SCHED_FIFO 或 SCHED_RR的优先级进程内可用。（The PTHREAD_PRIO_INHERIT and PTHREAD_PRIO_PROTECT mutex attributes are usable only by privileged processes running in the realtime (RT) scheduling class SCHED_FIFO or SCHED_RR.）
一个线程可以同时拥有多个混合使用 PTHREAD_PRIO_INHERIT 和 PTHREAD_PRIO_PROTECT协议属性初始化的互斥锁。在这种情况下,该线程将以通过其中任一协议获取的最高优先级执行。
pthread_mutexattr_getprotocol可用来获取互斥锁属性对象的协议属性。

优先级上限属性

int pthread_mutexattr_setprioceiling(pthread_mutexatt_t *attr, int prioceiling, int *oldceiling); //成功返回0，其它返回值表示出错  
int pthread_mutexattr_getprioceiling(const pthread_mutexatt_t *attr, int *prioceiling); //成功返回0，其它返回值表示出错

pthread_mutex_setprioceiling可更改互斥锁 mutex 的优先级上限 prioceiling。
pthread_mutex_setprioceiling可锁定互斥锁(如果未锁定的话),或者一直处于阻塞状态,直到它成功锁定该互斥锁,更改该互斥锁的优先级上限并将该互斥锁释放为止。锁定互斥锁的过程无需遵循优先级保护协议。如果 pthread_mutex_setprioceiling成功,则将在 old_ceiling 中返回以前的优先级上限值。如果pthread_mutex_setprioceiling失败,则互斥锁的优先级上限保持不变。
pthread_mutex_getprioceiling会返回 mutex 的优先级上限 prioceiling。

Robust 属性

TODO: 这个没有在工作中见过，不是特别了解，需要实验

int pthread_mutexattr_setrobust_np(pthread_mutexattr_t *attr, int *robustness);  //成功返回0，其它返回值表示出错 
int pthread_mutexattr_getrobust_np(const pthread_mutexattr_t *attr, int *robustness);  //成功返回0，其它返回值表示出错

pthread_mutexattr_setrobust_np用来设置互斥锁属性对象的强健属性。仅当定义了符号 _POSIX_THREAD_PRIO_INHERIT 时,pthread_mutexattr_setrobust_np()才适用。robustness 定义在互斥锁的持有者“死亡”时的行为。pthread.h 中定义的 robustness 的值为PTHREAD_MUTEX_ROBUST_NP 或 PTHREAD_MUTEX_STALLED_NP。缺省值为PTHREAD_MUTEX_STALLED_NP。

• PTHREAD_MUTEX_STALLED_NP： 如果互斥锁的持有者死亡,则以后对 pthread_mutex_lock() 的所有调用将以不确定的方式被阻塞。
• PTHREAD_MUTEX_ROBUST_NP： 如果互斥锁的持有者“死亡”了，或者持有这样的互斥锁的进程unmap了互斥锁所在的共享内存或者持有这样的互斥锁的进程执行了exec调用，则会解除锁定该互斥锁。互斥锁的下一个持有者将获取该互斥锁,并返回错误 EOWNWERDEAD。如果互斥锁具有PTHREAD_MUTEX_ROBUST_NP的属性，则应用程序在获取该锁时必须检查 pthread_mutex_lock 的返回代码看获取锁时是否返回了EOWNWERDEAD错误。如果是，则互斥锁的新的持有者应使该互斥锁所保护的状态保持一致。因为互斥锁的上一个持有者“死亡”时互斥锁所保护的状态可能出于不一致的状态。如果互斥锁的新的持有者能够使该状态保持一致,请针对该互斥锁调用pthread_mutex_consistent_np(),并解除锁定该互斥锁。如果互斥锁的新的持有者无法使该状态保持一致,请勿针对该互斥锁调用pthread_mutex_consistent_np(),而是解除锁定该互斥锁。所有等待的线程都将被唤醒,以后对 pthread_mutex_lock() 的所有调用都将无法获取该互斥锁。返回错误为ENOTRECOVERABLE。
  如果一个线程获取了互斥锁，但是获取时得到了EOWNERDEAD的错误，然后它终止并且没有释放互斥锁 ,则下一个持有者获取该锁时将返回代码EOWNERDEAD。

Note:

1. 互斥量需要时间来加锁和解锁。锁住较少互斥量的程序通常运行得更快。所以，互斥量应该尽量少，够用即可，每个互斥量保护的区域应则尽量大。
2. 互斥量的本质是串行执行。如果很多线程需要领繁地加锁同一个互斥量，则线程的大部分时间就会在等待，这对性能是有害的。如果互斥量保护的数据(或代码)包含彼此无关的片段，则可以特大的互斥量分解为几个小的互斥量来提高性能。这样，任意时刻需要小互斥量的线程减少，线程等待时间就会减少。所以，互斥量应该足够多(到有意义的地步)，每个互斥量保护的区域则应尽量的少。

Example

#include <pthread.h>
#include <cstdio>
#include <cstdlib>

struct ThreadData {
  int tid;
  int data;
};

int shared_x;
pthread_mutex_t lock;

void *ThreadProc(void *param) {
  ThreadData *data = static_cast<ThreadData *>(param);
  printf("begin from thread id: %d\n", data->tid);
  pthread_mutex_lock(&lock);
  shared_x += data->data;
  printf("thread %d: x = %d\n", data->tid, shared_x);
  pthread_mutex_unlock(&lock);
  pthread_exit(NULL);
}

int main(int argc, char *argv[]) {
  const int kNumThreads = 4;
  pthread_t threads[kNumThreads];
  ThreadData threads_data[kNumThreads];
  pthread_attr_t attr;

  shared_x = 0;
  pthread_mutex_init(&lock, NULL);
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
  for (int i = 0; i < kNumThreads; ++i) {
    threads_data[i].tid = i;
    threads_data[i].data = i * i;
    int rt = pthread_create(&threads[i], &attr, ThreadProc,
                            static_cast<void *>(&threads_data[i]));
    if (rt) {
      printf("ERROR: pthread_create failed, rt=%d\n", rt);
      exit(1);
    }
  }
  for (int i = 0; i < kNumThreads; ++i) {
    void *status;
    pthread_join(threads[i], &status);
  }
  pthread_attr_destroy(&attr);
  pthread_exit(NULL);
  return 0;
}

Condition Variables(条件变量)

Overview

Mutex 变量如锁一般防止多个线程访问共享数据资源,如果某个线程等待某个共享数据达到某个数值才进行相应的操作,那么这个线程需要不断的去 poll,查看是否满足需要的值,这样开销很大,因为线程需要一直处于忙状态.
引入 Condition Variables 来完成这样的同步到某个实际数据值而不要不断 poll.
Condition 变量一般与 mutex 一起使用.锁住查看的共享数据资源.
使用 Condition 的一般步骤如下:

• 声明和定义需要同步的共享数据;
• 声明和定义 condition 变量;
• 声明和定义相对应的 mutex;
• 创建线程使用 condition 变量同步.

条件变量创建与销毁

int pthread_cond_destroy(pthread_cond_t *cond);
int pthread_cond_init(pthread_cond_t *restrict cond,
                      const pthread_condattr_t *restrict attr);
int pthread_condattr_destroy(pthread_condattr_t *attr);
int pthread_condattr_init(pthread_condattr_t *attr);

Condition 变量由 pthread_cond_t 声明定义,而且必须初始化在使用前.两种方法初始:
静态的,当声明时.如:
pthread_cond_t convar = PTHREAD_COND_INITIALIZER;
动态的, 使用 pthread_cond_init() 函数,并能设置 condition 的属性 attr.

pthread_cond_t cond;
pthread_cond_init(& cond, nullptr);

attr 用来设置 condition 变量的属性,必须是 pthread_condattr_t 类型.只有一种属性可选:是否进程共享,也就是允许其他进程中的线程也能看到它.

pthread_condattr_t  cattr;
int ret;
/* initialize an attribute to default value */
ret = pthread_condattr_init(&cattr); 
/* all processes */
ret = pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_SHARED);
/* within a process */
ret = pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_PRIVATE);

cattr is an opaque data type that contains a system-allocated attribute object. The possible values of cattr's scope are PTHREAD_PROCESS_PRIVATE (the default) and PTHREAD_PROCESS_SHARED.

条件变量等待与信号

int pthread_cond_wait(pthread_cond_t *cond,
                      pthread_mutex_t *mutex);
int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);

pthread_cond_wait() 阻塞调用它的线程直到其中 cond 被 signal.这个函数需要在占有 mutex 时被调用,而当等待时它将自动释放 mutex.等到 signal 收到,线程被唤醒, mutex 将自动被占有 .最后当线程完成 condition 的操作,要负责对 mutex 解锁.
pthread_cond_signal() 用来 signal 其他等待这个 cond 的线程.它需要在占有 mutex 时被调用.然后必须对 mutex 解锁来完成 pthread_cond_wait 的等待.
如果有多于一个线程处于等待 cond 而阻塞, 应该用 pthread_cond_broadcast() 替换 pthread_cond_signal().

Example

#include <pthread.h>
#include <cstdio>
#include <cstdlib>
#include <unistd.h>

const int kNumThreads = 3;
const int kLoops = 10;
const int kCountLimit = 15;

int g_count;
pthread_mutex_t count_mutex;
pthread_cond_t count_cv;

void *IncreaseCount(void *param) {
  int id;
  id = *(static_cast<int *>(param));
  for (int i = 0; i < kLoops; ++i) {
    pthread_mutex_lock(&count_mutex);
    g_count++;
    if (g_count == kCountLimit) {
      pthread_cond_signal(&count_cv);
      printf("increse thread %d: count = %d, signal cond\n", id, g_count);
    }
    printf("increse thread %d: count = %d, unlock mutex\n", id, g_count);
    pthread_mutex_unlock(&count_mutex);
    sleep(1);
  }
  pthread_exit(NULL);
}

void *WatchCount(void *param) {
  int id;
  id = *(static_cast<int *>(param));
  pthread_mutex_lock(&count_mutex);
  while (g_count < kCountLimit) {
    pthread_cond_wait(&count_cv, &count_mutex);
    printf("watch thread %d: count = %d, receive signal\n", id, g_count);
  }
  pthread_mutex_unlock(&count_mutex);
  pthread_exit(NULL);
}

int main(int argc, char *argv[]) {
  pthread_t threads[kNumThreads];
  int thread_ids[kNumThreads];
  pthread_attr_t attr;

  pthread_mutex_init(&count_mutex, NULL);
  pthread_cond_init(&count_cv, NULL);
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
  for (int i = 0; i < kNumThreads; ++i) {
    thread_ids[i] = i;
  }
  int rt;
  rt = pthread_create(&threads[0], &attr, WatchCount,
                            static_cast<void *>(&thread_ids[0]));
  if (rt) {
    printf("ERROR: pthread_create failed, rt=%d\n", rt);
    exit(1);
  }
  rt = pthread_create(&threads[1], &attr, IncreaseCount,
                            static_cast<void *>(&thread_ids[1]));
  if (rt) {
    printf("ERROR: pthread_create failed, rt=%d\n", rt);
    exit(1);
  }
  rt = pthread_create(&threads[2], &attr, IncreaseCount,
                            static_cast<void *>(&thread_ids[2]));
  if (rt) {
    printf("ERROR: pthread_create failed, rt=%d\n", rt);
    exit(1);
  }
  for (int i = 0; i < kNumThreads; ++i) {
    pthread_join(threads[i], NULL);
  }
  pthread_attr_destroy(&attr);
  pthread_cond_destroy(&count_cv);
  pthread_mutex_destroy(&count_mutex);
  pthread_exit(NULL);
}

Semaphore

OverView

信号量本质上是一个非负的整数计数器，它被用来控制对公共资源的访问，也被称为PV原子操作。PV原子操作，广泛用于进程或线程之间的通信的同步和互斥。其中，P是通过的意思，V是释放的意思，不可中断的过程，则由操作系统来保证P操作和V操作。PV操作时针对信号量的操作，就是对信号量进行加减的过程。P操作，即信号量sem减一的过程，如果sem小于等于0，P操作被堵塞，直到sem变量大于0为止。P操作即加锁过程。V操作，即信号量sem加一的过程。V操作即解锁过程。

API

int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_wait(sem_t *sem);  // P操作，减少信号量
int sem_post(sem_t *sem);  //V操作， 增加信号量
int sem_destory(sem_t *sem);  //销毁信号量
int sem_getvalue(sem_t *sem, int *sval);  // 获取信号量的值

sem_init 函数说明：sem参数是信号量指针；pshared参数为共享方式，0表示信号量只是在当前进程中使用（线程），1表示信号量在多进程中使用；value参数表示信号量的初始值，一般为1。

Barrier

Overview

Barrier 就是栅栏一样,调用等待 barrier 的线程需要等待直到满足调用 barrier 的线程个数达到要求的 count.

API

int pthread_barrier_init(pthread_barrier_t *barrier,
                const pthread_barrierattr_t *attr, unsigned count);
pthread_barrier_t barrier = PTHREAD_BARRIER_INITIALIZER(count);
int pthread_barrier_destroy(pthread_barrier_t *barrier);
int pthread_barrierattr_init(pthread_barrierattr_t *attr);
int pthread_barrierattr_destroy(pthread_barrierattr_t *attr);
int pthread_barrier_wait(pthread_barrier_t *barrier);

The pthread_barrier_wait() function shall synchronize participating threads at the barrier referenced by barrier. The calling thread shall block until the required number of threads have called pthread_barrier_wait() specifying the barrier.
Barrier 变量由 pthread_barrier_t 声明定义,而且必须初始化在使用前.需要传入满足 barrier 等待的个数 count, 两种方法初始:
静态的,当声明时.如:
pthread_barrier_t barrier = PTHREAD_BARRIER_INITIALIZER(count);
动态的,使用 pthread_barrier_init() 函数,并能设置 barrier 的属性 attr.
线程调用 barrier,只需要调用 pthread_barrier_wait 来等待 barrier 达到满足条件.

Example

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <time.h>

#define THREAD_COUNT 4

pthread_barrier_t mybarrier;

void* threadFn(void *id_ptr) {
  int thread_id = *(int*)id_ptr;
  int wait_sec = 1 + rand() % 5;
  printf("thread %d: Wait for %d seconds.\n", thread_id, wait_sec);
  sleep(wait_sec);
  printf("thread %d: I'm ready...\n", thread_id);

  pthread_barrier_wait(&mybarrier);

  printf("thread %d: going!\n", thread_id);
  return NULL;
}


int main() {
  int i;
  pthread_t ids[THREAD_COUNT];
  int short_ids[THREAD_COUNT];

  srand(time(NULL));
  pthread_barrier_init(&mybarrier, NULL, THREAD_COUNT + 1);

  for (i=0; i < THREAD_COUNT; i++) {
    short_ids[i] = i;
    pthread_create(&ids[i], NULL, threadFn, &short_ids[i]);
  }

  printf("main() is ready.\n");

  pthread_barrier_wait(&mybarrier);

  printf("main() is going!\n");

  for (i=0; i < THREAD_COUNT; i++) {
    pthread_join(ids[i], NULL);
  }

  pthread_barrier_destroy(&mybarrier);

  return 0;
}

Reference:

浅谈C++ Multithreading Programming
freebsd pthreads manual page