Linux线程间交互

  • Linux线程间交互已关闭评论
  • 16 次浏览
  • A+
所属分类:linux技术
摘要

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


前言

上一篇说过,系统会为线程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 

加锁和解锁

  1. 使用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 

读写锁

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

  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 <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环境编程从应用到内核》高峰 李彬 著