Linux线程间交互

前言

上一篇说过，系统会为线程mmap一块内存，每个线程有自己的私有栈，使用局部变量没啥问题。但是实际场景中不可避免的需要线程之间共享数据，这就需要确保每个线程看到的数据是一样的，如果大家都只需要读这块数据没有问题，但是当有了修改共享区域的需求时就会出现数据不一致的问题。甚至线程2的任务在执行到某个地方的时候，需要线程1先做好准备工作，出现顺序依赖的情况。为了解决这些问题，Linux提供了多种API来适用于不同的场景。

互斥量 mutex

排他的访问共享数据，锁竞争激烈的场景使用。锁竞争不激烈的情况可以使用自旋锁（忙等）

当我们用trace -f 去追踪多线程的时候会看到执行加锁解锁的调用是futex，glibc通过futex（fast user space mutex）实现互斥量。通过FUTEX_WAIT_PRIVATE标志的futex调用内核的futex_wait挂起线程，通过FUTEX_WAKE_PRIVATE的futex调用内核的futex_wake来唤醒等待的线程。这之中glibc做了优化：

• 加锁时，当前mutex没有被加锁，则直接加锁，不做系统调用，自然不需要做上下文切换。如果已经加锁则需要系统调用futex_wait让内核将线程挂起到等待队列
• 解锁时，没有其他线程在等待该mutex，直接解锁，不做系统调用。如果有其他线程在等待，则通过系统调用futex_wake唤醒等待队列中的一个线程

初始化互斥量

#include 
// 动态初始化并设置互斥量属性，用完需要销毁
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
                       const pthread_mutexattr_t *restrict attr);
// attr 设置mutex的属性，NULL为使用默认属性
// 返回值：成功返回0，失败返回错误编号

// 静态初始化，无需销毁
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

销毁互斥量

// 销毁互斥量
int pthread_mutex_destroy(pthread_mutex_t *mutex);
// 返回值：成功返回0，失败返回错误编号。
// 		如果互斥量是锁定状态，或者正在和条件变量共同使用，销毁会返回EBUSY

加锁和解锁

1. 使用pthread_mutex_lock加锁

#include 
// 阻塞
int pthread_mutex_lock(pthread_mutex_t *mutex);
// 返回值：成功返回0，失败返回错误编号

// 非阻塞
int pthread_mutex_trylock(pthread_mutex_t *mutex);
// 返回值：加锁成功直接返回0，加锁失败返回EBUSY

int pthread_mutex_unlock(pthread_mutex_t *mutex);
// 返回值：成功返回0，失败返回错误编号

调用状态：

• 调用时互斥量未锁定，该函数所在线程争取到mutex，返回。
• 调用时已有其他线程对mutex加锁，则阻塞等待mutex被释放后重新尝试加锁

重复调用问题，即本线程已经对mutex加锁，再次调用加锁操作时，根据互斥量的类型不同会有不同表现：

• PTHREAD_MUTEX_TIMED_NP：重复加锁导致死锁，该调用线程永久阻塞，并且其他线程无法申请到该mutex
• PTHREAD_MUTEX_ERRORCHECK_NP：内部记录着调用线程，重复加锁返回EDEADLK，如果解锁的线程不是锁记录的线程，返回EPERM
• PTHREAD_MUTEX_RECURSIVE_NP：允许重复加锁，锁内部维护着引用计数和调用线程。如果解锁的线程不是锁记录的线程，返回EPERM
• PTHREAD_MUTEX_ADAPTIVE_NP（自适应锁）：先自旋一段时间，自旋的时间由__spins和MAX_ADAPTIVE_COUNT共同决定，自动调整__spin的大小但是不会超过MAX_ADAPTIVE_COUNT。超过自旋时间让出CPU等待，比自旋锁温柔，比normal mutex激进。

设置mutex属性

// 设置mutex为ADAPTER模式
pthread_mutexattr_t mutexattr;
pthread_mutexattr_init(&mutexattr);
pthread_mutexattr_settype(&mutexattr, PTHREAD_MUTEX_ADAPTIVE_NP);

// 获取mutex模式
int kind;
pthread_mutexattr_gettype(&mutexattr, &kind);
if (kind == PTHREAD_MUTEX_ADAPTIVE_NP) {
printf("mutex type is %s", "PTHREAD_MUTEX_ADAPTIVE_NP\n");
}

带有超时的mutex

int pthread_mutex_timedlock(pthread_mutex_t *restrict mutex, const struct timespec *restrict abstime);
// abstime表示在该时间之前阻塞，不是时间间隔
// 成功返回0，失败返回错误编号，超时返回ETIMIEOUT

demo

对已经加锁的mutex继续使用timedlock加锁，timedlock超时返回，之后mutex解锁

#define _DEFAULT_SOURCE 1
#include 
#include 
#include 
#include 
#include 
#include 

char* now_time(char buf[]) {
  struct timespec abstime;
  abstime.tv_sec = time(0);
  strftime(buf, 1024, "%r", localtime(&abstime.tv_sec));
  return buf;
}

int main() {
  char buf[1024];
  pthread_mutex_t mutex;
  struct timespec abstime;
  pthread_mutex_init(&mutex, NULL);
  pthread_mutex_lock(&mutex);
  char* now = now_time(buf);
  printf("mutex locked, now: %s\n", buf);
  // 设置超时的绝对时间，不设置tv_nsec会返回22，EINVAL
  abstime.tv_sec = time(0) + 10;
  abstime.tv_nsec = 0;
  int ret = pthread_mutex_timedlock(&mutex, &abstime);
  fprintf(stderr, "error %d\n", ret);
  if (ret == ETIMEDOUT) {
    printf("lock mutex timeout\n");
  } else if (ret == 0) {
    printf("lock mutex successfully\n");
  } else if (ret == EINVAL) {
    printf("timedlock param invalid!\n");
  } else {
    printf("other error\n");
  }
  pthread_mutex_unlock(&mutex);
  memset(buf, '\0', 1024);
  now = now_time(buf);
  printf("mutex unlocked, now: %s\n", buf);
  pthread_mutex_destroy(&mutex);
  return 0;
}

// -----------------------------
root@yielde:~/workspace/code-container/cpp/blog_demo# ./test 
mutex locked, now: 08:18:34 PM
error 110
lock mutex timeout
mutex unlocked, now: 08:18:44 PM

读写锁

读写锁适用于临界区很大并且在大多数情况下读取共享资源，极少数情况下需要写的场景

1. 未加锁：加读、写锁都可以
2. 加读锁：再次尝试加读锁成功，写锁阻塞
3. 加写锁：再次尝试加读、写锁阻塞

常用接口与mutex类似，用的时候查https://man7.org/linux/man-pages/dir_section_3.html，读写锁有两种策略：

PTHREAD_RWLOCK_PREFER_READER_NP, // 读者优先
PTHREAD_RWLOCK_PREFER_WRITER_NP, // 读者优先
PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP, // 写者优先
PTHREAD_RWLOCK_DEFAULT_NP = PTHREAD_RWLOCK_PREFER_READER_NP

// 通过以下函数设置
int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
int pthread_rwlockattr_getkind_np(const pthread_rwlockattr_t *attr, int *pref);

读写锁存在的问题：

1. 如果临界区小，锁内部维护的数据结构多于mutex，性能不如mutex
2. 因为有读优先和写优先的策略，使用不当会出现读或写线程饿死的现象
3. 如果是写策略优先，线程1持有读锁，线程2等待加写锁，线程1再次加读锁，就出现了死锁情况

demo

启动5个线程共同对一个变量累加1，使用读写锁让线程并发，用自适应锁对共享变量加锁。

/*
  5个线程对total加1执行指定次数
*/

#define _DEFAULT_SOURCE 1  // 处理vscode 未定义 pthread_rwlock_t
#include 
#include 
#include 
#include 

#define THREAD_COUNT 5

int total = 0;                                      // 最终和
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;  // 初始化互斥量
pthread_rwlock_t rwlock;                            // 读写锁变量
typedef struct param {                              // 线程参数类型
  int count;
  int id;
} param;

void *handler(void *arg) {
  struct param *pa = (struct param *)arg;
  pthread_rwlock_rdlock(&rwlock);  // 当主线程不unlock写锁时，会阻塞在这里
  for (int i = 0; i < pa->count; ++i) {
    pthread_mutex_lock(&mutex);  // 加互斥锁
    ++total;
    pthread_mutex_unlock(&mutex);
  }
  pthread_rwlock_unlock(&rwlock);
  printf("thread %d complete\n", pa->id);
  return NULL;
}

int main(int argc, char *argv[]) {
  if (argc != 2) {
    printf("usage: %s per_thread_loop_count\n", argv[0]);
    return 1;
  }
  // 设置mutex为ADAPTER模式
  pthread_mutexattr_t mutexattr;
  pthread_mutexattr_init(&mutexattr);
  pthread_mutexattr_settype(&mutexattr, PTHREAD_MUTEX_ADAPTIVE_NP);
  // 给handler传参
  int loop_count = atoi(argv[1]);
  // 存放线程id的数组
  pthread_t tid[THREAD_COUNT];
  param pa[THREAD_COUNT];

  pthread_rwlock_init(&rwlock, NULL);  // 动态初始化读写锁
  pthread_rwlock_wrlock(&rwlock);  // 给写加锁，等所有线程创建好后解锁，线程执行
  for (int i = 0; i < THREAD_COUNT; ++i) {  // 创建5个线程
    pa[i].count = loop_count;
    pa[i].id = i;
    pthread_create(&tid[i], NULL, handler, &pa[i]);
  }

  pthread_rwlock_unlock(&rwlock);
  for (int i = 0; i < THREAD_COUNT; ++i) {
    pthread_join(tid[i], NULL);
  }
  pthread_rwlock_destroy(&rwlock);
  printf("thread count: %d\n", THREAD_COUNT);
  printf("per thread loop count: %d\n", loop_count);
  printf("total except: %d\n", loop_count * 5);
  printf("total result: %d\n", total);

  int kind;
  pthread_mutexattr_gettype(&mutexattr, &kind);
  if (kind == PTHREAD_MUTEX_ADAPTIVE_NP) {
    printf("mutex type is %s", "PTHREAD_MUTEX_ADAPTIVE_NP\n");
  }
  return 0;
}

// --------------------------------
root@yielde:~/workspace/code-container/cpp/blog_demo# ./test 2000
thread 2 complete
thread 1 complete
thread 0 complete
thread 3 complete
thread 4 complete
thread count: 5
per thread loop count: 2000
total except: 10000
total result: 10000
mutex type is PTHREAD_MUTEX_ADAPTIVE_NP

自旋锁

等待锁的时候不会通知内个将线程挂起，而是忙等。适用于临界区很小，锁被持有的时间很短的情况，相比于互斥锁，节省了上下文切换的开销

线程同步-屏障

barrier可以同步多个线程，允许任意数量的线程等待，直到所有的线程完成工作，然后继续执行

#include 

int pthread_barrier_destroy(pthread_barrier_t *barrier);
// 返回值：成功返回0，失败返回错误号
int pthread_barrier_init(pthread_barrier_t *restrict barrier,
   	const pthread_barrierattr_t *restrict attr, unsigned count);
// count指定有多少个线程到达屏障后再继续执行下去
// 返回值：成功返回0，失败返回错误号

int pthread_barrier_wait(pthread_barrier_t *barrier);
// 成功：给一个线程返回PTHREAD_BARRIER_SERIAL_THREAD，其他线程返回0
// 失败返回错误号

demo

使用4个线程，每个线程计算1+1+..+1=10，将结果放入数组的一个位置，完成后到达barrier。主线程创建好线程后到达barrier，等四个线程全部完成后，由主线程合计结果

#define _DEFAULT_SOURCE
#include 
#include 
#include 
#define COUNT 10
#define THR_NUM 4

pthread_barrier_t barrier;
long total_arr[THR_NUM] = {0};

void *handler(void *arg) {
  long idx = (long)arg;
  long tmp = 0;
  for (int i = 0; i < COUNT; ++i) {
    ++tmp;
    sleep(1);
  }
  total_arr[idx] = tmp;
  printf("thread %ld complete, count %ld\n", idx, tmp);
  pthread_barrier_wait(&barrier); // 等待在barrier
  return NULL;
}

int main() {
  pthread_t tids[THR_NUM];
  unsigned long total = 0;

  pthread_barrier_init(&barrier, NULL, THR_NUM + 1);  // 包含主线程
  for (long i = 0; i < THR_NUM; ++i) {
    pthread_create(&tids[i], NULL, handler, (void *)i);
  }
  pthread_barrier_wait(&barrier); // 到达barrier
  for (int i = 0; i < THR_NUM; ++i) {
    total += total_arr[i];
  }

  for (int i = 0; i < THR_NUM; ++i) {
    pthread_join(tids[i], NULL);
  }
  pthread_barrier_destroy(&barrier); // 销毁barrier
  printf("total: %lu\n", total);
}

// ---------------------
root@yielde:~/workspace/code-container/cpp/blog_demo# time ./test
thread 2 complete, count 10
thread 0 complete, count 10
thread 3 complete, count 10
thread 1 complete, count 10
total: 40

real    0m10.027s
user    0m0.005s
sys     0m0.003s

线程同步-条件变量

如果条件不满足，线程会等待在条件变量上，并且让出mutex，等待其他线程来执行。其他线程执行到条件满足后会发信号唤醒等待的线程。

// 销毁条件变量
int pthread_cond_destroy(pthread_cond_t *cond);

// 初始化条件变量
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);

// 等待条件变量
int pthread_cond_timedwait(pthread_cond_t *restrict cond,
   		pthread_mutex_t *restrict mutex,
   		const struct timespec *restrict abstime);
int pthread_cond_wait(pthread_cond_t *restrict cond,
   		pthread_mutex_t *restrict mutex);

// 通知条件变量满足
int pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒所有线程
int pthread_cond_signal(pthread_cond_t *cond); // 至少唤醒1个线程
//返回值成功返回0，失败返回错误号

对于 cond_wait，传递mutex保护条件变量，调用线程将锁住的mutex传给函数，函数将调用线程挂起到等待队列上，解锁互斥量。当函数返回时，互斥量再次被锁住。

demo

handler_hello往buf里输入字符串，由handler_print打印

#include 
#include 
#include 
#include 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; // 初始化互斥量
pthread_cond_t cond = PTHREAD_COND_INITIALIZER; // 初始化条件变量

char buf[8] = {0};

void *handler_hello(void *arg) {
  for (;;) {
    sleep(2);
    pthread_mutex_lock(&mutex);
    sprintf(buf, "%s", "hello !");
    pthread_mutex_unlock(&mutex);
    pthread_cond_signal(&cond); // 唤醒wait的线程
  }

  return NULL;
}

void *handler_print(void *arg) {
  for (;;) {
    pthread_mutex_lock(&mutex);
    while (buf[0] == 0) {
        // 如果buf没有内容就等待，此处将线程挂入队列，然后解锁mutex，等收到handler_hello的signal后返回，加锁mutex
        // 
      pthread_cond_wait(&cond, &mutex); 
    }
    fprintf(stderr, "%s", buf);
    memset(buf, '\0', 8);
    pthread_mutex_unlock(&mutex);
  }
  return NULL;
}

int main() {
  pthread_t tid1, tid2;
  pthread_create(&tid1, NULL, handler_hello, NULL);
  pthread_create(&tid2, NULL, handler_print, NULL);

  pthread_join(tid1, NULL);
  pthread_join(tid2, NULL);

  printf("%s", buf);
  return 0;
}


// ------------------------
root@yielde:~/workspace/code-container/cpp/blog_demo# ./test
hello !hello !hello !hello !^C

学习自：
《UNIX环境高级编程》
《Linux环境编程从应用到内核》高峰 李彬 著