- A+
所属分类：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_NPn"); } 带有超时的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: %sn", 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 %dn", ret); if (ret == ETIMEDOUT) { printf("lock mutex timeoutn"); } else if (ret == 0) { printf("lock mutex successfullyn"); } else if (ret == EINVAL) { printf("timedlock param invalid!n"); } else { printf("other errorn"); } pthread_mutex_unlock(&mutex); memset(buf, ' ', 1024); now = now_time(buf); printf("mutex unlocked, now: %sn", 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 completen", pa->id); return NULL; } int main(int argc, char *argv[]) { if (argc != 2) { printf("usage: %s per_thread_loop_countn", 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: %dn", THREAD_COUNT); printf("per thread loop count: %dn", loop_count); printf("total except: %dn", loop_count * 5); printf("total result: %dn", total); int kind; pthread_mutexattr_gettype(&mutexattr, &kind); if (kind == PTHREAD_MUTEX_ADAPTIVE_NP) { printf("mutex type is %s", "PTHREAD_MUTEX_ADAPTIVE_NPn"); } 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 %ldn", 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: %lun", 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, ' ', 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环境编程从应用到内核》高峰 李彬 著
