OS76.【Linux】线程互斥(5) 同步

假设有一间VIP自习室,最多只能一个人自习,该自习室有个规则: 先到先得,VIP自习室有人就上锁(其他人无法进入,相当于保护VIP自习室这个共享资源),VIP自习室无人就开锁(相当于释放共享资源)

任何人来到无人的VIP自习室后都需要上锁(人相当于执行流,这是为了保证只有一个执行流访问共享资源)

显然任何执行流访问共享资源前都要能够访问到这个锁; 执行流先访问锁,如果成功获取锁,那么才能真正的抢到票,否则线程可能会因为加不上锁导致阻塞

如果自习室中的人结束自习后,又因为后悔释放了空位,他可以立马再次申请VIP自习室空位,进而反复占用资源,导致其它人无法上自习

为了防止VIP自习室空位分配不合理的问题,定2条规则:

1.外面来的人,必须排队

2.出来的人,不能立马重新申请自习室空位,必须排到队列的尾部

2.同步

VIP自习室2条规则对应到锁资源上就是:

1.线程想要申请锁,必须排队

2.释放锁的线程,不能立马重新申请锁,必须排到队列的尾部

定义

同步的定义: 保证资源安全的情况下,让所有的执行流按照一定的顺序获取资源

反过来思考: 线程都排队了,已经按照一定顺序访问了,为什么需要加锁?

要明白为什么去排队,是因为线程访问资源失败了,才会去排队,而且随时随地都会有线程访问资源,因此必须保证资源的安全,"保证资源安全的情况下"前提很重要

条件变量

定义

纯互斥方案是能解决一些问题的,例如多线程均衡访问资源,但纯互斥有局限性,纯互斥没有同步,不保证顺序,所以可能出现调度不均衡的问题 → 比如上面提到的"如果自习室中的人结束自习后,又因为后悔释放了空位,他可以立马再次申请VIP自习室空位,进而反复占用资源,导致其它人无法上自习"

条件变量(Condition Variable)可以实现同步

假设线程抢到锁之后, 发现没有可利用的资源,就会在条件变量上去等待

比如在生产者-消费者模型中(这个后面会讲):

消费者线程抢到锁之后, 发现没有可消费的资源,就会在消费者条件变量上去等待

同理,生产者抢到锁之后, 发现没有可生产的资源,就会在生产者条件变量上去等待

当占用锁的线程释放锁时,如果资源就绪,操作系统会唤醒(敲铃铛)等待队列中的线程,唤醒方式一般分两种:

1.可以唤醒一个

2.也可以唤醒多个

选择唤醒方式,具体看程序员的想法,被唤醒的线程可以选择申请锁,也可以退出,视线程的具体情况而定

这里线程获取的资源是加了锁的,加锁可以保证资源的安全,当线程申请锁失败了才会排队,所以条件变量必须配合互斥锁

结论: 条件变量=铃铛(用于唤醒执行流)+等待队列; 条件变量必须配合锁的使用

pthread_cond_init函数

int pthread_cond_init(pthread_cond_t *restrict cond,
           const pthread_condattr_t *restrict attr);

作用: 初始化条件变量,pthread_cond_init的cond是condition的缩写

参数cond是条件变量指针

参数attr是条件变量的属性,一般写nullptr,表示默认

如果条件变量定义为全局变量,那么直接使用PTHREAD_COND_INITIALIZER宏,不需要初始化函数和销毁函数

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

pthread_cond_destroy函数

int pthread_cond_destroy(pthread_cond_t *cond);

作用: 顾名思义,销毁条件变量,

手册里面写的很清楚,这里不是将cond指向的条件变量从内存中除去,而是将条件变量修改为未初始化的状态(destroy the given condition variable specified by cond; the object becomes, in effect, uninitialized.)

pthread_cond_wait函数

int pthread_cond_wait(pthread_cond_t *restrict cond,
           pthread_mutex_t *restrict mutex);

作用: 顾名思义,如果等待条件满足,将线程放到等待队列中休眠

参数cond: 要在这个条件变量上等待

参数mutex: 互斥量

pthread_cond_signal和pthread_cond_broadcast函数

int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);

这两个函数是唤醒方式

pthread_cond_signal是唤醒在这个条件变量下的等待队列中的一个线程,一般是队列中的第一个,但pthread_cond_broadcast是唤醒在这个条件变量下的等待队列中的所有线程

注意: "在这个条件变量下"是指cond指向的条件变量

代码演示: 多线程轮流修改全局变量cnt

多线程轮流对全局变量cnt做+1操作,但是一个线程只加一次

先看看多线程得到数字的情况:

#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <iostream>
#define NUM 5
int cnt=0;
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
void* thread_routine(void* args)
{
    uint64_t num=(uint64_t)args;
    for (;;)
    {
        pthread_mutex_lock(&lock);
        printf("thread-%ld cnt=%d\n",num,++cnt);
        pthread_mutex_unlock(&lock);
    }
    return nullptr;
}

int main()
{
    pthread_t tid_arr[NUM];
    for (uint64_t i=0;i<NUM;i++)
    {
        //不能取地址i,主线程执行循环,&i会让新线程和主线程的i一样,Linux多线程不能保证主线程和新线程的调度顺序
        //这里(void*)i拷贝不会出问题,这样主线程和新线程不会相互影响
        pthread_create(&tid_arr[i],nullptr,thread_routine,(void*)i);
    }
    for (int i=0;i<NUM;i++)
    {
        pthread_join(tid_arr[i],nullptr);
    }
    return 0;
}

运行结果:每次输出的结果都不一样

注意: 不能写成这样,原因上面代码的注释说过了

void* thread_routine(void* args)
{
    uint64_t num=*(uint64_t*)args;
    printf("%ld\n",num);
    return nullptr;
}

//main内
for (uint64_t i=0;i<NUM;i++)
    pthread_create(&tid_arr[i],nullptr,thread_routine,(void*)&i);

运行结果:

由于多线程轮流对全局变量cnt做加法,那么需要对cnt加锁,简单起见,这里加全局锁

#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <iostream>
#define NUM 5
int cnt=0;
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
void* thread_routine(void* args)
{
    uint64_t num=(uint64_t)args;
    for (;;)
    {
        pthread_mutex_lock(&lock);
        printf("thread-%ld cnt=%d\n",num,++cnt);
        pthread_mutex_unlock(&lock);
    }
    return nullptr;
}

int main()
{
    pthread_t tid_arr[NUM];
    for (uint64_t i=0;i<NUM;i++)
    {
        pthread_create(&tid_arr[i],nullptr,thread_routine,(void*)i);
    }
    for (int i=0;i<NUM;i++)
    {
        pthread_join(tid_arr[i],nullptr);
    }
    return 0;
}

但这样写有问题:pthread_mutex_unlock执行后,一轮循环结束,下一次循环立刻执行pthread_mutex_lock,导致锁被一个线程长时间持有

解决方法: 同步

为了防止锁被一个线程长时间持有,使用条件变量

做法: 这里为了简单起见,不管临界资源是否充足,只要有进程得到锁就放到等待队列中,可以调用pthread_mutex_lock(&lock)后立刻调用pthread_cond_wait(&cond,&lock)

,之后主线程依次调用pthread_cond_signal唤醒线程

pthread_cond_wait是将线程放到条件变量下的等待队列中排队,pthread_cond_signal是唤醒等待队列中的其中一个线程,默认是第一个

#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <iostream>
#define NUM 5
int cnt=0;
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
void* thread_routine(void* args)
{
    pthread_detach(pthread_self());
    uint64_t num=(uint64_t)args;
    for (;;)
    {
        pthread_mutex_lock(&lock);
        pthread_cond_wait(&cond,&lock);
        printf("thread-%ld cnt=%d\n",num,++cnt);
        pthread_mutex_unlock(&lock);
    }
    return nullptr;
}

int main()
{
    pthread_t tid_arr[NUM];
    for (uint64_t i=0;i<NUM;i++)
    {
        pthread_create(&tid_arr[i],nullptr,thread_routine,(void*)i);
    }
    for (;;)
    {
        pthread_cond_signal(&cond);
        sleep(1);
    }
    return 0;
}

运行结果: 除了第一行,其余线程打印满足一定的顺序

★为什么是先pthread_mutex_lock(&lock)后pthread_cond_wait(&cond,&lock)?

所有线程都来争锁,但是当前只有一个全局锁,假设线程0执行pthread_mutex_lock(&lock)得到了锁,

其下一步执行pthread_cond_wait(&cond,&lock),pthread_cond_wait(&cond,&lock)会自动将锁释放

原因:

pthread_cond_wait会自动先释放锁再让线程到队列中等待,如果pthread_cond_wait没有自动释放锁,而是让线程持有锁到等待队列中排队的话,其它线程会因为抢不到锁而继续阻塞等锁,那么会出现死锁的情况,所有线程都将无法继续执行!!!

线程0执行pthread_cond_wait后,将锁释放出来,那么其它线程就能继续取得锁,这样所有的线程100%有机会在等待队列中休眠,等待主线程一个个唤醒即可,而且pthread_cond_signal需要知道唤醒哪个条件变量下的等待队列中线程,pthread_cond_signal会唤醒线程,被唤醒线程从pthread_cond_wait返回时,pthread_cond_wait函数内部会自动重新获取之前传入的互斥锁,因为唤醒的线程需要访问临界资源

注意: 一定是先调用pthread_mutex_lock(&lock)后调用pthread_cond_wait(&cond,&lock)

,因为是以线程为单位执行thread_routine,线程必须先取得锁,才能用pthread_cond_wait释放锁

当然也可以按照0~4的顺序唤醒,让主线程睡眠一段时间即可:

#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <iostream>
#define NUM 5
int cnt=0;
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
void* thread_routine(void* args)
{
    pthread_detach(pthread_self());
    uint64_t num=(uint64_t)args;
    for (;;)
    {
        pthread_mutex_lock(&lock);
        pthread_cond_wait(&cond,&lock);
        printf("thread-%ld cnt=%d\n",num,++cnt);
        pthread_mutex_unlock(&lock);
    }
    return nullptr;
}

int main()
{
    pthread_t tid_arr[NUM];
    for (uint64_t i=0;i<NUM;i++)
    {
        pthread_create(&tid_arr[i],nullptr,thread_routine,(void*)i);
        usleep(1000);
    }
    for (;;)
    {
        pthread_cond_signal(&cond);
        sleep(1);
    }
    return 0;
}

运行结果:

如果主线程改用pthread_cond_broadcast唤醒线程,运行结果:

源码验证: pthread_cond_wait自动加锁和解锁

在glibc-2.43的/nptl/pthread_cond_wait.c中定义了pthread_cond_wait函数:

pthread_cond_wait和___pthread_cond_wait是等同的

/* See __pthread_cond_wait_common.  */
int
___pthread_cond_wait (pthread_cond_t *cond, pthread_mutex_t *mutex)
{
  /* clockid is unused when abstime is NULL. */
  return __pthread_cond_wait_common (cond, mutex, 0, NULL);
}

versioned_symbol (libc, ___pthread_cond_wait, pthread_cond_wait,
		  GLIBC_2_3_2);

___pthread_cond_wait内部调用__pthread_cond_wait_common ,也定义在/nptl/pthread_cond_wait.c中:

static __always_inline int
__pthread_cond_wait_common (pthread_cond_t *cond, pthread_mutex_t *mutex,
    clockid_t clockid, const struct __timespec64 *abstime)
{
  int err;
  int result = 0;

  LIBC_PROBE (cond_wait, 2, cond, mutex);

  err = __pthread_mutex_unlock_usercnt (mutex, 0);
  if (__glibc_unlikely (err != 0))
    {
      __condvar_cancel_waiting (cond, seq, g, private);
      __condvar_confirm_wakeup (cond, private);
      return err;
    }


  while (1)
    {
      unsigned int signals = atomic_load_acquire (cond->__data.__g_signals + g);
      uint64_t g1_start = __condvar_load_g1_start_relaxed (cond);

      if (seq < g1_start)
        {
           break;
        }

      if ((int)(signals - (unsigned int)g1_start) > 0)
        {
	  /* Try to grab a signal.  See above for MO.  (if we do another loop
	     iteration we need to see the correct value of g1_start)  */
	    if (atomic_compare_exchange_weak_acquire (
			cond->__data.__g_signals + g,
			&signals, signals - 1))
	      break;
	    else
	      continue;
	}

      struct _pthread_cleanup_buffer buffer;
      struct _condvar_cleanup_buffer cbuffer;
      cbuffer.wseq = wseq;
      cbuffer.cond = cond;
      cbuffer.mutex = mutex;
      cbuffer.private = private;
      __pthread_cleanup_push (&buffer, __condvar_cleanup_waiting, &cbuffer);

      err = __futex_abstimed_wait_cancelable64 (
        cond->__data.__g_signals + g, signals, clockid, abstime, private);

      __pthread_cleanup_pop (&buffer, 0);

      if (__glibc_unlikely (err == ETIMEDOUT || err == EOVERFLOW))
        {
          __condvar_cancel_waiting (cond, seq, g, private);
          result = err;
          break;
        }
    }
  __condvar_confirm_wakeup (cond, private);

  err = __pthread_mutex_cond_lock (mutex);
  return (err != 0) ? err : result;
}

该函数一开始就释放锁(__pthread_mutex_unlock_usercnt),结束时重新取得锁(__pthread_mutex_cond_lock):

static __always_inline int
__pthread_cond_wait_common (pthread_cond_t *cond, pthread_mutex_t *mutex,
    clockid_t clockid, const struct __timespec64 *abstime)
{
  int err;
  int result = 0;

  LIBC_PROBE (cond_wait, 2, cond, mutex);

  err = __pthread_mutex_unlock_usercnt (mutex, 0);
  //......

  err = __pthread_mutex_cond_lock (mutex);
  return (err != 0) ? err : result;
}

__pthread_mutex_unlock_usercnt 和__pthread_mutex_cond_lock 第一个参数都是同一个mutex

结论: pthread_cond_wait释放的锁和获取的锁都是同一个锁