Android C++ 系列：Linux 线程（四）线程同步

2021 年 12 月 07 日
本文字数：4848 字
阅读完需：约 16 分钟

多个线程同时访问共享数据时可能会冲突，这跟我们前面信号文章所说的可重入性是同样的问题。比如两个线程都要把某个全局变量增加 1，这个操作在某平台需要三条指令完成:

从内存读变量值到寄存器；
寄存器的值加 1；
将寄存器的值写回内存

假设两个线程在多处理器平台上同时执行这三条指令，则可能导致下图所示的结果，最后变量只加了一次而非两次。

实例：

#include <stdio.h> #include <stdlib.h> #include <pthread.h>#define NLOOP 5000 int counter;void *doit(void *);/* incremented by threads */int main(int argc, char **argv) {  pthread_t tidA, tidB;   pthread_create(&tidA, NULL, &doit, NULL);  pthread_create(&tidB, NULL, &doit, NULL);  /* wait for both threads to terminate */   pthread_join(tidA, NULL);  pthread_join(tidB, NULL);  return 0; }void *doit(void *vptr) {  int i, val;  for (i = 0; i < NLOOP; i++) {     val = counter;    printf("%x: %d\n", (unsigned int)pthread_self(), val + 1);    counter = val + 1;   }  return NULL; }

复制代码

我们创建两个线程，各自把 counter 增加 5000 次，正常情况下最后 counter 应该等于 10000，但事实上每次运行该程序的结果都不一样，有时候数到 5000 多，有时候数到 6000 多。

1. 线程为什么要同步

共享资源，多个线程都可对共享资源操作；
线程操作共享资源的先后顺序不确定；
处理器对存储器的操作一般不是原子操作。

2. 互斥量

mutex 操作原语：

pthread_mutex_t
pthread_mutex_init
pthread_mutex_destroy
pthread_mutex_lock
pthread_mutex_trylock
pthread_mutex_unlock

2.1 临界区（Critical Section）

保证在某一时刻只有一个线程能访问数据的简便办法。在任意时刻只允许一个线程对共享资源进行访问。如果有多个线程试图同时访问临界区，那么在有一个线程进入后其他所有试图访问此临界区的线程将被挂起，并一直持续到进入临界区的线程离开。临界区在被释放后，其他线程可以继续抢占，并以此达到用原子方式操作共享资源的目的。

2.2 临界区的选定

临界区的选定因尽可能小，如果选定太大会影响程序的并行处理性能。

2.3 互斥量实例

#include <pthread.h>pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;int pthread_mutex_destroy(pthread_mutex_t *mutex);int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr); int pthread_mutex_lock(pthread_mutex_t *mutex);int pthread_mutex_trylock(pthread_mutex_t *mutex);int pthread_mutex_unlock(pthread_mutex_t *mutex);

复制代码

实例：

#include <stdio.h> #include <stdlib.h> #include <pthread.h> #define NLOOP 5000int counter; /* incremented by threads */ pthread_mutex_t counter_mutex = PTHREAD_MUTEX_INITIALIZER;void *doit(void *);int main(int argc, char **argv) {  pthread_t tidA, tidB;  pthread_create(&tidA, NULL, doit, NULL);   pthread_create(&tidB, NULL, doit, NULL);  /* wait for both threads to terminate */   pthread_join(tidA, NULL);  pthread_join(tidB, NULL);  return 0; }void *doit(void *vptr) {  int i, val;  for (i = 0; i < NLOOP; i++) {     pthread_mutex_lock(&counter_mutex);    val = counter;    printf("%x: %d\n", (unsigned int)pthread_self(), val + 1);     counter = val + 1;    pthread_mutex_unlock(&counter_mutex);   }  return NULL; }

复制代码

这样运行结果就正常了，每次运行都能数到 10000。

3. 死锁

同一个线程在拥有 A 锁的情况下再次请求获得 A 锁；
线程一拥有 A 锁，请求获得 B 锁;线程二拥有 B 锁，请求获得 A 锁死锁导致的结果是什么?

4. 读写锁

读共享，写独占

pthread_rwlock_t
pthread_rwlock_init
pthread_rwlock_destroy
pthread_rwlock_rdlock
pthread_rwlock_wrlock
pthread_rwlock_tryrdlock
pthread_rwlock_trywrlock
pthread_rwlock_unlock

实例：

#include <stdio.h>#include <pthread.h>int counter;pthread_rwlock_t rwlock; //3个线程不定时写同一全局资源，5个线程不定时读同一全局资源 void *th_write(void *arg){  int t;  while (1) {     pthread_rwlock_wrlock(&rwlock);     t = counter;    usleep(100);    printf("write %x : counter=%d ++counter=%d\n", (int)pthread_self(), t, ++counter);    pthread_rwlock_unlock(&rwlock);     usleep(100);  } }void *th_read(void *arg) {  while (1) {    pthread_rwlock_rdlock(&rwlock);    printf("read %x : %d\n", (int)pthread_self(), counter);           pthread_rwlock_unlock(&rwlock);    usleep(100);  } }int main(void) {  int i;  pthread_t tid[8];   pthread_rwlock_init(&rwlock, NULL);   for (i = 0; i < 3; i++)    pthread_create(&tid[i], NULL, th_write, NULL);   for (i = 0; i < 5; i++)    pthread_create(&tid[i+3], NULL, th_read, NULL);  pthread_rwlock_destroy(&rwlock);  for (i = 0; i < 8; i++)    pthread_join(tid[i], NULL);   return 0;}

复制代码

5. 条件变量

条件变量给多个线程提供了一个汇合的场所,条件变量控制原语:

pthread_cond_t
pthread_cond_init
pthread_cond_destroy
pthread_cond_wait
pthread_cond_timedwait
pthread_cond_signal
pthread_cond_broadcast

生产者消费者模型:

#include <stdlib.h> #include <pthread.h> #include <stdio.h>struct msg {  struct msg *next;   int num;};struct msg *head;pthread_cond_t has_product = PTHREAD_COND_INITIALIZER; pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;void *consumer(void *p) {  struct msg *mp;  for (;;) {   pthread_mutex_lock(&lock);   while (head == NULL)    pthread_cond_wait(&has_product, &lock);   mp = head;  head = mp->next;   pthread_mutex_unlock(&lock);   printf("Consume %d\n", mp->num); free(mp);  sleep(rand() % 5);  } }void *producer(void *p) {  struct msg *mp;   for (;;) {    mp = malloc(sizeof(struct msg));     mp->num = rand() % 1000 + 1;     printf("Produce %d\n", mp->num);     pthread_mutex_lock(&lock);     mp->next = head;    head = mp;     pthread_mutex_unlock(&lock);     pthread_cond_signal(&has_product);     sleep(rand() % 5);  } }int main(int argc, char *argv[]) {  pthread_t pid, cid;  srand(time(NULL));  pthread_create(&pid, NULL, producer, NULL);   pthread_create(&cid, NULL, consumer, NULL);   pthread_join(pid, NULL);  pthread_join(cid, NULL);  return 0;}

复制代码

6. 信号量

信号量控制原语

sem_t
sem_init
sem_wait
sem_trywait
sem_timedwait
sem_post
sem_destroy

生产者消费者实例：

#include <stdlib.h> #include <pthread.h> #include <stdio.h> #include <semaphore.h>#define NUM 5int queue[NUM];sem_t blank_number, product_number;void *producer(void *arg) {  int p = 0;   while (1) {    sem_wait(&blank_number);    queue[p] = rand() % 1000 + 1;     printf("Produce %d\n", queue[p]);     sem_post(&product_number);    p = (p+1)%NUM;    sleep(rand()%5);  } }void *consumer(void *arg) {  int c = 0;   while (1) {    sem_wait(&product_number);     printf("Consume %d\n", queue[c]); queue[c] = 0; sem_post(&blank_number);    c = (c+1)%NUM;    sleep(rand()%5);   }}int main(int argc, char *argv[]) {  pthread_t pid, cid;  sem_init(&blank_number, 0, NUM);  sem_init(&product_number, 0, 0);   pthread_create(&pid, NULL, producer, NULL);   pthread_create(&cid, NULL, consumer, NULL);  pthread_join(pid, NULL);   pthread_join(cid, NULL);   sem_destroy(&blank_number);   sem_destroy(&product_number);   return 0;}

复制代码

7. 进程间锁

7.1 进程间 pthread_mutex

#include <pthread.h>int pthread_mutexattr_init(pthread_mutexattr_t *attr);int pthread_mutexattr_setpshared(pthread_mutexattr_t *attr, int pshared); int pthread_mutexattr_destroy(pthread_mutexattr_t *attr);

复制代码

pshared:

线程锁:PTHREAD_PROCESS_PRIVATE ;
进程锁:PTHREAD_PROCESS_SHARED；
默认情况是线程锁

实例：

#include <stdio.h> #include <pthread.h> #include <unistd.h> #include <sys/stat.h> #include <sys/types.h> #include <fcntl.h> #include <sys/mman.h> #include <string.h>struct mt {   int num;  pthread_mutex_t mutex;  pthread_mutexattr_t mutexattr; };int main(void) {  int fd, i;  struct mt *mm;  pid_t pid;  fd = open("mt_test", O_CREAT | O_RDWR, 0777);  /* 不需要write,文件里初始值为0 */  ftruncate(fd, sizeof(*mm));  mm = mmap(NULL, sizeof(*mm), PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);   close(fd);  memset(mm, 0, sizeof(*mm));  /* 初始化互斥对象属性 */   pthread_mutexattr_init(&mm->mutexattr);  /* 设置互斥对象为PTHREAD_PROCESS_SHARED共享，即可以在多个进程的线程访问,PTHREAD_PROCESS_PRIVATE 为同一进程的线程共享 */  pthread_mutexattr_setpshared(&mm->mutexattr,PTHREAD_PROCESS_SHARED);  pthread_mutex_init(&mm->mutex, &mm->mutexattr);  pid = fork();   if (pid == 0){    /* 加10次。相当于加10 */     for (i=0;i<10;i++){      pthread_mutex_lock(&mm->mutex);       (mm->num)++;       printf("num++:%d\n",mm->num);       pthread_mutex_unlock(&mm->mutex);       sleep(1);    }   }else if (pid > 0) {    /* 父进程完成x+2,加10次，相当于加20 */     for (i=0; i<10; i++){      pthread_mutex_lock(&mm->mutex);       mm->num += 2;       printf("num+=2:%d\n",mm->num);       pthread_mutex_unlock(&mm->mutex);       sleep(1);    }    wait(NULL);   }  pthread_mutex_destroy(&mm->mutex);   pthread_mutexattr_destroy(&mm->mutexattr);   /* 父子均需要释放 */   munmap(mm,sizeof(*mm));  unlink("mt_test");  return 0;}

复制代码

7.2 文件锁

使用 fcntl 提供文件锁

struct flock {   ...  short l_type; /*Type of lock: F_RDLCK, F_WRLCK, F_UNLCK */  short l_whece; /* How to interpret l_start: SEEK_SET,SEEK_CUR,SEEK_END */  off_t l_start; /* Starting offset for lock*/  off_t l_len; /*Number of bytes to lock*/  pid_t l_pid; /* PID of process blocking our lock (F_GETLK only) */

复制代码

实例：

#include <stdio.h> #include <sys/types.h> #include <sys/stat.h> #include <fcntl.h> #include <unistd.h> #include <stdlib.h> void sys_err(char *str) {  perror(str);  exit(1); }int main(int argc, char *argv[]) {  int fd;  struct flock f_lock;   if (argc < 2) {    printf("./a.out filename\n");    exit(1);   }  if ((fd = open(argv[1], O_RDWR)) < 0)   sys_err("open");  //f_lock.l_type = F_WRLCK;   f_lock.l_type = F_RDLCK;   f_lock.l_whence = SEEK_SET;   f_lock.l_start = 0;  f_lock.l_len = 0; //0表示整个文件加锁  fcntl(fd, F_SETLKW, &f_lock);   printf("get flock\n");   sleep(10);  f_lock.l_type = F_UNLCK;   fcntl(fd, F_SETLKW, &f_lock);   printf("un flock\n");  close(fd);  return 0; }

复制代码

8. 总结

本文介绍了线程同步机制：为什么要同步、互斥量、死锁、读写锁、条件变量、信号量、进程间锁等概念与机制以及相关示例。

发布于: 15 小时前阅读数: 6

原文链接:【http://xie.infoq.cn/article/b0f9f32057f49ced328bb0353】。文章转载请联系作者。

轻口味

关注

🏆2021年InfoQ写作平台-签约作者 🏆 2017.10.17 加入

Android、音视频、AI相关领域从业者。邮箱：qingkouwei@gmail.com

发布

暂无评论

创作场景

Android C++ 系列：Linux 线程（四）线程同步

1. 线程为什么要同步

2. 互斥量

2.1 临界区（Critical Section）

2.2 临界区的选定

2.3 互斥量实例

3. 死锁

4. 读写锁

5. 条件变量

6. 信号量

7. 进程间锁

7.1 进程间 pthread_mutex

7.2 文件锁

8. 总结

轻口味

评论