Android C++系列：Linux线程线程同步

Posted 2022-03-29 轻口味丿

​

多个线程同时访问共享数据时可能会冲突，这跟我们前面信号文章所说的可重入性是同样的问题。比如两个线程都要把某个全局变量增加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. 线程为什么要同步

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

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 5000
int 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. 死锁

1. 同一个线程在拥有A锁的情况下再次请求获得A锁；
2. 线程一拥有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 5
int 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. 总结

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