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 <pthread.h>
// 动态初始化并设置互斥量属性，用完需要销毁
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
加锁和解锁
- 使用pthread_mutex_lock加锁
#include <pthread.h>
// 阻塞
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 <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
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
读写锁
读写锁适用于临界区很大并且在大多数情况下读取共享资源，极少数情况下需要写的场景
- 未加锁：加读、写锁都可以
- 加读锁：再次尝试加读锁成功，写锁阻塞
- 加写锁：再次尝试加读、写锁阻塞
常用接口与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);
读写锁存在的问题：
- 如果临界区小，锁内部维护的数据结构多于mutex，性能不如mutex
- 因为有读优先和写优先的策略，使用不当会出现读或写线程饿死的现象
- 如果是写策略优先，线程1持有读锁，线程2等待加写锁，线程1再次加读锁，就出现了死锁情况
demo
启动5个线程共同对一个变量累加1，使用读写锁让线程并发，用自适应锁对共享变量加锁。
/*
5个线程对total加1执行指定次数
*/
#define _DEFAULT_SOURCE 1 // 处理vscode 未定义 pthread_rwlock_t
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#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 <pthread.h>
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 <pthread.h>
#include <stdio.h>
#include <unistd.h>
#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 <pthread.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
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环境编程从应用到内核》高峰 李彬 著