Linux锁概述&互斥锁+条件变量实现生产者与消费者模式

前言

在使用锁时需要明确的点：

  • 临界区域(critical section)是指多使用者可能同时共同操作的那部分代码，比如自加自减操作，多个线程处理时就需要对自加自减进行保护，这段代码就是临界区域。
  • 锁的所有权问题，谁加锁，谁解锁，这就是解铃还须系铃人。
  • 锁的作用就是对临界区资源的读写操作的安全限制。

学习锁要带着下面几个问题：

  • 锁是否可以被多个使用者占用（互不影响的使用者对资源的占用）；
  • 占用资源的加锁者的释放问题（锁持有的超时问题）；
  • 等待资源的待加锁者的等待问题（如何通知到其他等着资源的使用者）；
  • 多个临界区资源锁的循环问题（死锁场景）。

互斥锁

互斥锁是对临界区资源进行锁定，保证同一时刻只能有一个线程去操作。
互斥锁是多个线程一起去抢，抢到锁的线程先执行，没有抢到锁的线程需要等待，等互斥锁使用完释放后，其它等待的线程再去抢这个锁。

自旋锁

是为实现保护临界区资源而提出一种锁机制。其实，自旋锁与互斥锁比较类似，它们都是为了解决对某项资源的互斥使用。无论是互斥锁，还是自旋锁，在任何时刻，最多只能有一个保持者，也就说，在任何时刻最多只能有一个执行单元获得锁。但是两者在调度机制上略有不同。对于互斥锁，如果资源已经被占用，资源申请者只能进入睡眠状态。但是自旋锁不会引起调用者睡眠，如果自旋锁已经被别的执行单元保持，调用者就一直循环在那里看是否该自旋锁的保持者已经释放了锁，"自旋"一词就是因此而得名。

自旋锁并不会放弃CPU时间片，而是通过自旋等待锁的释放，也就是说，它会不停地再次地尝试获取锁，如果失败就再次尝试，直到成功为止。
如果在临界区资源代码非常短且是原子的，那么使用起来是非常方便的，避免了各种上下文切换，开销非常小，因此在内核的一些数据结构中自旋锁被广泛的使用。

对比互斥锁和自旋锁

自旋转自己原地旋转来确定锁被释放了，区别于自旋锁，互斥锁无法获取锁时将阻塞睡眠，需要系统来唤醒。但是互斥锁在某些业务场景中无法借助系统来唤醒，仍然需要业务代码使用while来判断，这种效率是比较低的。我们可以配合条件变量来解决这个问题，具体内容将会在后文条件变量一节中展开。
值得注意的是互斥锁要进行上下文切换，自旋锁不用，而上下文切换很耗资源。

读写锁

一次只有一个线程可以占有写模式的读写锁, 但是可以有多个线程同时占有读模式的读写锁。
也就是只有三种情况：1、既没有读也没有写；2、多个线程在读，没有线程写；3、一个线程在写，没有线程读。
在读写锁保持期间也是抢占失效的。

  • 如果读写锁当前没有读者，也没有写者，那么写者可以立刻获得读写锁，否则它必须自旋在那里，直到没有任何写者或读者。
  • 如果读写锁没有写者，那么读者可以立即获得该读写锁，否则读者必须自旋在那里，直到写者释放该读写锁。

读写锁适合于对数据结构的读次数比写次数多得多的情况。因为, 读模式锁定时可以共享, 以写模式锁住时意味着独占, 所以读写锁又叫共享-独占锁。

RCU锁

RCU锁（Read copy update lock，读-拷贝-更新 锁）
RCU锁是读写锁的扩展版本，简单来说就是支持多读多写同时加锁。

  • 读(Read)：读者不需要获得任何锁就可访问RCU保护的临界区；
  • 拷贝(Copy)：写者在访问临界区时，写者“自己”将先拷贝一个临界区副本，然后对副本进行修改；
  • 更新(Update)：RCU机制将在在适当时机使用一个回调函数把指向原来临界区的指针重新指向新的被修改的临界区，锁机制中的垃圾收集器负责回调函数的调用。(时机：所有引用该临界区资源的CPU都退出对临界区的操作。即没有CPU再去操作这段被RCU保护的临界区后，这段临界区即可回收了，此时回调函数即被调用)

从实现逻辑来看，RCU锁在多个写者之间的同步开销还是比较大的，涉及到多份数据拷贝，回调函数等，因此这种锁机制的使用范围比较窄，适用于读多写少的情况，如网络路由表的查询更新、设备状态表更新等，在业务开发中使用不是很多。

可重入锁和不可重入锁

Windows下的Mutex和Critical Section是可递归的。Linux下的pthread_mutex_t锁默认是非递归的。在Linux中可以显式设置PTHREAD_MUTEX_RECURSIVE属性，将pthread_mutex_t设为递归锁避免这种场景。 同一个线程可以多次获取同一个递归锁，不会产生死锁。而如果一个线程多次获取同一个非递归锁，则会产生死锁。

同步与互斥的区别

  • 同步与互斥机制是用于控制多个任务对某些特定资源的访问策略；
  • 同步是控制多个任务按照一定的规则或顺序访问某些共享资源；
  • 互斥是控制某些共享资源在任意时刻只能允许规定数量的任务访问。

其中同步是由条件变量来实现。

条件变量

  • 条件变量用于等待。条件变量通常搭配互斥锁来使用， 是因为条件的检测是在互斥锁的保护下进行的， 也就是说条件本身是由互斥锁保护的， 线程在改变条件状态之前必须首先锁住互斥锁， 不然就可能引发线程不安全的问题。
  • 从另一个角度来说，条件变量通过允许线程阻塞和等待另一个线程发送信号来弥补互斥锁的不足，所以互斥锁和条件变量通常一起使用，来让条件变量异步唤醒阻塞的线程。

总结就是，条件变量需要在互斥锁的保护下工作，互斥锁需要条件变量来弥补不能异步唤醒的不足。
如果多个线程在锁着，条件变量唤醒其中一个线程，其他线程就是虚假唤醒，如果不想虚假唤醒可以使用broadcast全部唤醒。

经典的生产者消费者模式，便可以由条件变量和互斥锁来实现。
下面是生产者消费者的示例代码：

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

constexpr int MAX_PRODUCE_COUNT = 1 << 4;
constexpr int MAX_QUEUE_SIZE = 1 << 2;
int g_current_produce_index = 0;
std::queue<int> g_pool;

pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t g_cond_produce = PTHREAD_COND_INITIALIZER;
pthread_cond_t g_cond_consume = PTHREAD_COND_INITIALIZER;

class lock_guard
{   // class with destructor that unlocks a mutex
public:
   explicit lock_guard(pthread_mutex_t& mutex)
       :m_mutex(mutex)
   {   // construct and lock
       pthread_mutex_lock(&m_mutex);
   }


   ~lock_guard() noexcept
   {   // unlock
       pthread_mutex_unlock(&m_mutex);
   }

   lock_guard(const lock_guard&) = delete;
   lock_guard& operator=(const lock_guard&) = delete;
private:
   pthread_mutex_t& m_mutex;
};

void* produce(void* p_thread_no)
{
   int thread_no = *(int*)p_thread_no;
   while(true)
   {
       lock_guard lock(g_mutex);
       while (g_pool.size() >= MAX_QUEUE_SIZE)
       {
           pthread_cond_wait(&g_cond_produce, &g_mutex);
       }
       // do produce
       g_pool.push(g_current_produce_index);
       printf("produce %d in thread_%d\n", g_current_produce_index++, thread_no);
       pthread_cond_signal(&g_cond_consume);
       sleep(0.01);
       if (g_current_produce_index >= MAX_PRODUCE_COUNT)
           pthread_exit((void*)1);
   }
}

void* consume(void* p_thread_no)
{
   int thread_no = *(int*)p_thread_no;
   while (true)
   {
       lock_guard lock(g_mutex);
       while (g_pool.empty())
       {
           pthread_cond_wait(&g_cond_consume, &g_mutex);
       }
       // do consume
       printf("consume %d in thread_%d\n", g_pool.front(), thread_no);
       g_pool.pop();
       pthread_cond_signal(&g_cond_produce);
       sleep(0.01);
       if (g_current_produce_index >= MAX_PRODUCE_COUNT and g_pool.empty())
           pthread_exit((void*)2);
   }
}

int main()
{
   pthread_t thread_id_1;
   pthread_t thread_id_2;
   pthread_t thread_id_3;
   pthread_t thread_id_4;

   int thread_1 = 1;
   int thread_2 = 2;
   int thread_3 = 3;
   int thread_4 = 4;

   pthread_create(&thread_id_1, NULL, produce, &thread_1);
   pthread_create(&thread_id_2, NULL, produce, &thread_2);
   pthread_create(&thread_id_3, NULL, consume, &thread_3);
   pthread_create(&thread_id_4, NULL, consume, &thread_4);

   pthread_join(thread_id_1, NULL);
   pthread_join(thread_id_2, NULL);
   pthread_join(thread_id_3, NULL);
   pthread_join(thread_id_4, NULL);

   pthread_cond_destroy(&g_cond_produce);
   pthread_cond_destroy(&g_cond_consume);
   return 0;
}

~$ g++ consume_produce.cpp -lpthread
~$ ./a.out
produce 0 in thread_1
produce 1 in thread_1
produce 2 in thread_1
produce 3 in thread_1
consume 0 in thread_3
consume 1 in thread_3
consume 2 in thread_3
consume 3 in thread_3
produce 4 in thread_1
produce 5 in thread_1
produce 6 in thread_1
produce 7 in thread_1
consume 4 in thread_4
consume 5 in thread_4
consume 6 in thread_4
consume 7 in thread_4
produce 8 in thread_2
produce 9 in thread_2
produce 10 in thread_2
produce 11 in thread_2
consume 8 in thread_3
consume 9 in thread_3
consume 10 in thread_3
consume 11 in thread_3
produce 12 in thread_1
produce 13 in thread_1
produce 14 in thread_1
produce 15 in thread_1
consume 12 in thread_4
consume 13 in thread_4
consume 14 in thread_4
consume 15 in thread_4
produce 16 in thread_2
consume 16 in thread_3

信号量

信号量和互斥锁的场景不一样，信号量主要是资源数量的管理（池化）。信号量既可以用于进程也可以用于线程。
由于信号量只能进行两种操作等待和发送信号，即P(sv)和V(sv),他们的行为是这样的：

  • P(sv)：如果sv的值大于零，就给它减1；如果它的值为零，就挂起该进程的执行。
  • V(sv)：如果有其他进程因等待sv而被挂起，就让它恢复运行，如果没有进程因等待sv而挂起，就给它加1。

死锁

锁无法解除的同时也无法加持，导致程序可能会无限阻塞的情况称为死锁。
线程死锁：

  • 如果一个线程多次获取同一个非递归锁，则会产生死锁；
  • 多个线程对多个互斥锁交叉使用，每一个线程都试图对其他线程所持有的互斥锁进行加锁。

进程死锁：

  • 竞争不可抢占资源而引起死锁；
    比如进程a打开文件f1，进程b打开文件f2，然后进程a去打开f2，进程b去打开f1，f1和f2都被占有了，此时进程a和进程b就会无限等待下去，形成死锁。
  • 竞争可消耗资源而引起死锁 ；
    比如进程a产生消息m1发给进程b，进程b收到消息后发送消息m2给进程a，进程a要收到m2才能产生消息m1，就会造成进程a和进程b死锁。
    *进程推进顺序不当而引起死锁（进程运行过程中，请求和释放资源的顺序不当，而导致进程死锁）。
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容