【Linux】线程----线程安全

概念

• 多个执行流对临界资源进行争抢访问，而不会造成数据二义性或者逻辑混乱，称这段争抢访问的过程是线程安全的。

实现方法

线程安全的实现：

• 同步：通过条件判断，实现对临界资源访问的时序合理性
• 互斥：通过唯一访问，实现对临界资源访问的安全性

互斥

• 互斥的实现技术：互斥锁、信号量
• 实现互斥的原理：只要保证同一时间只有一个执行流能够访问资源就是互斥
• 对临界资源进行状态标记：没人访问的时候标记为1，表示可以访问；有人正在访问的时候，就标记为0，表示不可访问；在对临界资源进行访问之前先进行状态判断，决定是否能够访问，不能访问则使其休眠。

互斥锁

互斥锁原理

• 互斥锁：其实就是一个计数器，只有0/1的计数器，用于标记资源当前的访问状态
  1----可访问
  0----不可访问
• 互斥锁想要实现互斥，每个线程在访问临界资源之前都要先访问同一个互斥锁（加锁）；意味着互斥锁本身就是一个临界资源（涉及到计数器的修改，修改过程必须保证安全，如果连自己都保护不了，如何保护他人？）
  如果是普通的计数器，则操作步骤为 将mutex的值加载到CPU的一个寄存器，然后判断是否可以访问，最后返回数据，此操作不是原子的。
• 互斥锁的计数器操作如何实现原子性：
  示意图：
  

结论：

互斥锁操作流程：

1. 定义互斥锁

pthread_mutex_t mutex;

2. 初始化互斥锁

mutex=PTHREAD_MUTEX_INITIALIZER / 
int pthread_mutex_init(pthread_mutex_t *mutex, pthread_mutexattr_t *attr)
mutex：互斥锁变量首地址
attr：互斥锁属性，通常置NULL

3. 在对临界资源访问之前，先加锁（访问锁，判断是否可以访问）----保护对临界资源访问的过程

int pthread_mutex_lock(pthread_mutex_t *mutex);// 阻塞加锁----如果不能加锁，则一直等待
int pthread_mutex_trylock(pthread_mutex_t *mutex);// 非阻塞加锁----如果不能加锁，则立即报错返回；若可以加锁，则加锁后返回

4. 在对临界资源访问完毕之后，记得解锁（把状态标记为可访问）

int pthread_mutex_unlock(pthread_mutex_t *mutex);

5. 不使用锁了，最终要释放资源，销毁互斥锁

int pthread_mutex_destroy(pthread_mutex_t *mutex);

举例

以黄牛抢票为例：
不加锁的代码：

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

int g_tickets = 100; // 总共有100张票

void *thr_Scalpers(void *arg)
{
  while(1)
  {
    if(g_tickets > 0)
    {
      usleep(1000);
      printf("I : [%p] got a ticket : %d\n", pthread_self(), g_tickets);
      g_tickets--;
    }
    else 
    {
      pthread_exit(NULL); // 票抢完退出
    }
  }
  return NULL;
}

int main()
{
  int i = 0;
  pthread_t tid[4]; // 4个线程（黄牛）
  for(i = 0; i < 4; ++i)
  {
    int ret = pthread_create(&tid[i], NULL, thr_Scalpers, NULL);
    if(ret != 0)
    {
      printf("thread create error!\n");
      return -1;
    }
  }
  for(i = 0; i < 4; ++i)
  {
    pthread_join(tid[i], NULL);
  }
  return 0;
}

结果：

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

int g_tickets = 100;

// 1.定义互斥锁变量，并且这个变量也额是一个临界资源，需要能够被各个线程访问到
// 可以将互斥锁变量定义为一个全局变量，也可以将其定义成局部变量，然后通过函数传参传递给线程
pthread_mutex_t mutex;

void *thr_Scalpers(void *arg)
{
  // 尽量避免对不需要加锁的操作进行加锁，会影响效率，因为在加锁期间别人都不能操作
  // 若加锁的操作越多，意味着需要的时间就更长
  while(1)
  {
    pthread_mutex_lock(&mutex); // 加锁一定是在临界资源访问之前，保护的也仅仅是临界区
    if(g_tickets > 0)
    {
      usleep(1000);
      printf("I : [%p] got a ticket : %d\n", pthread_self(), g_tickets);
      g_tickets--;

      // 解锁是在临界资源访问完毕之后
      pthread_mutex_unlock(&mutex);
      usleep(1000); // 防止一个黄牛抢完马上又抢（一个线程分到时间片，解锁后立马又加锁）
    }
    else 
    {
      // 在任意有可能退出线程的地方，记得解锁
      // 若退出没有解锁，则其他线程获取不到锁，就会卡死
      pthread_mutex_unlock(&mutex);
      pthread_exit(NULL);
    }
  }
  return NULL;
}

int main()
{
  int i = 0;
  pthread_t tid[4];
  // 2.初始化互斥锁，一定要在创建线程之前！
  pthread_mutex_init(&mutex, NULL);

  for(i = 0; i < 4; ++i)
  {
    int ret = pthread_create(&tid[i], NULL, thr_Scalpers, NULL);
    if(ret != 0)
    {
      printf("thread create error!\n");
      return -1;
    }
  }
  for(i = 0; i < 4; ++i)
  {
    pthread_join(tid[i], NULL);
  }
  // 3.销毁互斥锁，释放资源
  pthread_mutex_destroy(&mutex);
  return 0;
}

结果：

1. 加锁保护的区域，最好只对临界资源的访问过程----保护的越多，执行的所需时间越长，效率越低
2. 在任意有可能退出线程的地方，记得解锁
3. 锁的初始化一定要在创建线程之前
4. 锁的销毁一定是保证没有人使用互斥锁的时候

死锁

死锁：多个执行流对锁资源进行争抢访问，但因为推进顺序不当，而导致相互等待，最终造成程序流程无法继续的情况

死锁产生的必要条件：

1. 互斥条件：一个资源每次只能被一个执行流使用
2. 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
3. 不剥夺条件:一个执行流已获得的资源，在末使用完之前，不能强行剥夺
4. 循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系

预防死锁

在编码过程中注意破坏死锁产生的必要条件

避免死锁

• 死锁检测算法
• 银行家算法

同步

• 通过条件判断（什么时候能访问资源，什么时候不能访问；若不能访问就要使线程阻塞；若能访问了就要唤醒线程），实现线程对临界资源访问的合理性
• 同步的实现技术：条件变量、信号量

条件变量

• 实现同步的思路向用户提供两个接口（一个是让线程陷入休眠的接口；一个是唤醒休眠线程的接口）+ pcb等待队列
• 条件变量只是向外提供了等待与唤醒的接口，却没有提供条件判断（条件变量本身并不具备判断什么时候该等待，什么时候该唤醒）的功能，意味着条件判断需要用户自己来完成

条件变量提供的接口功能

1. 定义条件变量

pthread_cond_t cond;

2. 初始化条件变量

pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *attr); / 
cond=PTHREAD_COND_INITIALIZER;

3. 一个线程等待的接口

pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);// 阻塞等待
pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, 
         struct timespec *abstime); //限制等待时长的阻塞操作：等待一段指定的时间，
                                    //时间到了调用就会报错返回----ETIMEOUT

• 条件变量是搭配互斥锁使用的：条件变量并不提供条件判断的功能，需要用户自己去判断（通常条件的判断是一个临界资源的访问），因此这个临界资源的访问就需要受保护，使用互斥锁保护。

4. 一个唤醒线程的接口

pthread_cond_signal(pthread_cond_t *cond); // 唤醒至少一个等待的线程
pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒所有等待的线程

5. 若不使用条件变量了则销毁释放资源

pthread_cond_destroy(pthread_cond_t *cond);

例子

以顾客和厨师为例子：

• 现在有一个顾客和一个厨师和一个碗
• 顾客：

0. 使用互斥锁加锁
1. 判断碗是否为空–满则吃饭/空则cond_wait，进入等待队列
   解锁
   挂起
   （被唤醒时）加锁
2. 吃饭
3. 唤醒厨师
4. 解锁

• 厨师：

0. 使用互斥锁加锁
1. 判断碗是否为空–空则做饭/满则cond_wait，进入等待队列
2. 做饭
3. 唤醒顾客pthread_cond_signal
4. 解锁

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

int bowl = 0; // 碗初始为0，表示没有饭

pthread_mutex_t mutex;
pthread_cond_t cond;

void *thr_customer(void *arg)
{
  // 这是一个顾客流程
  while(1)
  {
    // 0.加锁
    pthread_mutex_lock(&mutex);
    if(bowl == 0)
    {
      // 如果没有饭，则要等待,因为已经加过锁了，所以等待之前要解锁，被唤醒之后要加锁
      // 因此pthread_cond_wait集合了三步操作：解锁/挂起/加锁
      // 解锁和挂起是一个原子操作-不可被打断
      // 顾客解锁，还没来得及挂起休眠，这时候厨师来做饭，做好后唤醒顾客（实际顾客还没休眠）
      pthread_cond_wait(&cond, &mutex);
    }
    printf("good tast!\n");
    bowl = 0; // 饭被吃完
    // 唤醒厨师，再做一碗
    pthread_cond_signal(&cond);
    // 解锁操作
    pthread_mutex_unlock(&mutex);
  }
  return NULL;
}

void *thr_cook(void *arg)
{
  // 这是一个厨师流程
  while(1)
  {
    // 0.加锁操作，因为要对碗进行操作
    pthread_mutex_lock(&mutex);
    if(bowl == 1)
    {
      // 有饭，陷入等待
      pthread_cond_wait(&cond, &mutex);
    }
    printf("cook...\n");
    bowl = 1; // 做了一碗饭
    // 唤醒顾客
    pthread_cond_signal(&cond);
    // 解锁操作
    pthread_mutex_unlock(&mutex);
  }
  return NULL;
}

int main()
{
  pthread_t ctid[2];
  int ret;

  pthread_mutex_init(&mutex, NULL);
  pthread_cond_init(&cond, NULL);

  ret = pthread_create(&ctid[0], NULL, thr_customer, NULL);
  if(ret != 0)
  {
    printf("create customer failed!\n");
    return -1;
  }

  ret = pthread_create(&ctid[1], NULL, thr_cook, NULL);
  if(ret != 0)
  {
    printf("create cook failed!\n");
    return -1;
  }

  pthread_join(ctid[0], NULL);
  pthread_join(ctid[1], NULL);

  pthread_mutex_destroy(&mutex);
  pthread_cond_destroy(&cond);
  return 0;
}

问题分析

• 代码和逻辑在只有一个顾客和一个厨师的时候没有问题，但如果有多个顾客和多个厨师则会出现混乱。

分析：

1. 假设有三个顾客和三个厨师，顾客先进，由于碗空，三个顾客休眠进入等待队列；
2. 厨师A拿到了碗，另外两个厨师在锁上等待；
3. 厨师A做完饭，唤醒顾客吃饭并解锁；
4. 另外两个厨师发现碗里有饭也进入等待队列；
5. 假设顾客A和B被唤醒吃饭，此时只有顾客A能吃，顾客B在锁上等待；
6. 顾客A吃完后唤醒厨师，解锁；厨师被唤醒，但是不一定有时间片，解锁之后有可能顾客B抢到锁，此时碗里并没有饭，但顾客B依然吃饭了（吃了没有的饭） ，此时逻辑混乱。

所以判断碗是否有饭应该是一个循环判断，在有多个顾客的情况下，大家都争抢锁，拿到锁之后再重新判断一次有没有饭，有饭则不再等待，吃饭；没有饭则重新调用pthread_cond_wait休眠。

此时逻辑正确了，但还有问题

此时pcb等待队列中，既有顾客pcb也有厨师pcb；
顾客吃完饭后要去队列中唤醒厨师；
但大家在同一队列中，因此有可能唤醒的不是厨师，而是另一个顾客；唤醒角色错误，顾客因为没有饭休眠，程序卡死。

所以不同的角色应该等待在不同的队列中。

即程序中角色有多少种，条件变量就应该有多少个，各自等待在自己的队列中。

修改后的代码：

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

int bowl = 0; // 碗初始为0，表示没有饭

pthread_mutex_t mutex;
pthread_cond_t consumer_cond;
pthread_cond_t cook_cond;

void *thr_customer(void *arg)
{
  // 这是一个顾客流程
  while(1)
  {
    // 0.加锁
    pthread_mutex_lock(&mutex);
    while(bowl == 0)
    {
      // 如果没有饭，则要等待,因为已经加过锁了，所以等待之前要解锁，被唤醒之后要加锁
      // 因此pthread_cond_wait集合了三步操作：解锁/挂起/加锁
      // 解锁和挂起是一个原子操作-不可被打断
      // 顾客解锁，还没来得及挂起休眠，这时候厨师来做饭，做好后唤醒顾客（实际顾客还没休眠）
      pthread_cond_wait(&consumer_cond, &mutex);
    }
    printf("good tast!\n");
    bowl = 0; // 饭被吃完
    // 唤醒厨师，再做一碗
    pthread_cond_signal(&cook_cond);
    // 解锁操作
    pthread_mutex_unlock(&mutex);
  }
  return NULL;
}

void *thr_cook(void *arg)
{
  // 这是一个厨师流程
  while(1)
  {
    // 0.加锁操作，因为要对碗进行操作
    pthread_mutex_lock(&mutex);
    while(bowl == 1)
    {
      // 有饭，陷入等待
      pthread_cond_wait(&cook_cond, &mutex);
    }
    printf("cook...\n");
    bowl = 1; // 做了一碗饭
    // 唤醒顾客
    pthread_cond_signal(&consumer_cond);
    // 解锁操作
    pthread_mutex_unlock(&mutex);
  }
  return NULL;
}

int main()
{
  pthread_t ctid[2];
  int ret;
  int i = 0;

  pthread_mutex_init(&mutex, NULL);
  pthread_cond_init(&cook_cond, NULL);
  pthread_cond_init(&consumer_cond, NULL);

  for(i = 0; i < 4; ++i)
  {
    ret = pthread_create(&ctid[0], NULL, thr_customer, NULL);
    if(ret != 0)
    {
      printf("create customer failed!\n");
      return -1;
    }
  }


  for(i = 0; i < 4; ++i)
  {
    ret = pthread_create(&ctid[1], NULL, thr_cook, NULL);
    if(ret != 0)
    {
      printf("create cook failed!\n");
      return -1;
    }
  }

  pthread_join(ctid[0], NULL);
  pthread_join(ctid[1], NULL);

  pthread_mutex_destroy(&mutex);
  pthread_cond_destroy(&cook_cond);
  pthread_cond_destroy(&consumer_cond);
  return 0;
}

注意

1. pthread_cond_wait中包含了三步操作（解锁 + 休眠 + （被唤醒后）加锁，解锁和休眠是一个连在一起的原子操作）
2. 用户自己进行的条件判断需要使用while循环判断
3. 不同的角色应该等待在不同的条件变量上（有多少角色，就有多少条件变量）