C++ 实现同步基础组件

yfs

2022-08-22

C++, 基础架构

C++ 实现同步基础组件

1.Mutex

在muduo中,锁的实现包含了两个部分:Mutex和MutexGuard,它们都利用了RAII的机制,前者控制pthread_mutex的创建和销毁,后者控制一个pthread_mutex的上锁与解锁.

其实现大体如下:

class Mutex  : tinynetlib::nocopyable {
    public:
        Mutex() {
            pthread_mutex_init(&mutex_, NULL);
        }
        ~Mutex() {
            pthread_mutex_destroy(&mutex_);
        }
        void lock() { pthread_mutex_lock(&mutex_); }
        void unlock() { pthread_mutex_unlock(&mutex_); }

        pthread_mutex_t *getMutex() { return &mutex_; }

    private:
        pthread_mutex_t mutex_;
};

以上是对于Mutex的封装,其中构造函数为一个pthread_mutex_t的创建,析构函数为对于一个pthread_mutex_t的销毁.其中关于lock,unlock的实现并不是被用户直接调用的,而是间接地通过MutexGuard这个类的构造函数和析构函数调用.正如下所示:

class MutexLock :tinynetlib::nocopyable{
    public:
        explicit MutexLock(Mutex &mutex) : mutex_(mutex) {
            mutex_.lock();
        }
        ~MutexLock() {
            mutex_.unlock();
        }
    private:
        Mutex &mutex_;
    };

还有一个需要注意的地方在于,其中MutexLock这个类的成员数据是一个Mutex的引用,为什么是引用呢?这也更加体现了,这个类在于控制一个Mutex,而不是代表一个含有Mutex实体的类.也就是说,它并不产生一个新的Mutex对象,而是控制一个已有的Mutex对象.

正如下所示,这也几乎就是有关于锁的主要应用操作.

//......
{
     MutexLock lock(mutex_);
     isRunning = false;
     if(!tasksqueue_.empty()){
         cond_.signalAll();
}
//......

表示大括号内就是一个临界区.

2.Condition variable

条件变量最主要的设计初衷在于,能够以线程阻塞代替轮询,从而减少了大量的无效轮询,减少了CPU开销.但是有阻塞就必须要有唤醒,因此唤醒也是条件变量的设计的主要任务.

在应用上,条件变量常用于是某些未达到某个条件的线程阻塞,因而不会反复轮询这个条件.当这个条件为真时,唤醒某个睡眠的线程.

对于这个特定的条件,由于是需要多个线程进行读写的,因此也就需要一个Mutex用来防止对这个条件造成DATA RACE.此外,当进入等待状态时,也需要将这个Mutex释放,否则就会引发死锁.

其实现如下:

class Condition : tinynetlib::nocopyable{
    public:
        explicit Condition(Mutex &mutex):mutex_(mutex) {
            pthread_cond_init(&cond_,NULL);
        }
        ~Condition() {
            pthread_cond_destroy(&cond_);
        }
        void wait() {
            pthread_cond_wait(&cond_,mutex_.getMutex());
        }
        void signal() {
            pthread_cond_signal(&cond_);
        }
        void signalAll() {
            pthread_cond_broadcast(&cond_);
        }
    private:
        Mutex &mutex_;
        pthread_cond_t cond_;
    };

而对于关于.Condition variable的调用上,以线程池中的应用为例.首先是关于wait的应用,在这里条件变量所感兴趣的条件在于tasksqueue是否为空,如果为空就wait.在这里,为什么使用while而不是if呢?这个问题的原因称之为spurcious wakeup.

MutexLock lock(self->mutex_);
if(!self->isRunning) {
		return;
}
while(self->tasksqueue_.empty()) {
    self->cond_.wait();
}
task = self->tasksqueue_.front();
self->tasksqueue_.pop();

关于唤醒的调用:

MutexLock lock(mutex_);
		if (!isRunning) {
		return;
}
tasksqueue_.push(task);
cond_.signal();

表示此时该线程的行为表示这个条件有可能被改变,所以此时需要唤醒睡眠在上面的线程.

总的来说,对Mutex和Condition variable来说,都是同部分问题中最最基础的实现,一般都没有被用户调用,而是间接地实现在其他的类中.

3.Thread Pool

Thread Pool的实现也是基于Mutex和Condition variable实现的,其中这个数据结构维护一个队列用来记录线程池所接受到的任务.

其实线程池也是一个标准的“生产者-消费者模型”,线程池作为消费者,其中维护一个task队列,当需要处理一个任务时,就加入到线程池所维护的任务列表里,而线程池里的每个一个线程则需要检查队列的大小,如果是空的,就睡眠在条件变量上.否则,任务就会被唤醒.

不过在muduo中,还增加了一个条件变量,其对应的条件是,任务队列是否为满.

4.CountDownLatch

这个组件,最初我们并没有明白其意思,直到后来接触可golang之后开始明白其作用,其实现方式非常简单.

class CountDownLatch : noncopyable
{
 public:

  explicit CountDownLatch(int count);

  void wait();

  void countDown();

  int getCount() const;

 private:
  mutable MutexLock mutex_;
  Condition condition_ GUARDED_BY(mutex_);
  int count_ GUARDED_BY(mutex_);
};

其中对外开放的接口主要是wait,countDown等.其中的锁用来维护count_变量,条件变量通过读取以count作为条件控制调用者的阻塞.其实现更具体地说,再wait中就是while循环中判断是否大于0,如果是,就wait.countDown这个变量用来将变量-1,倘若到了0,则将所有正在在其条件变量上的线程唤醒.

所有总的来说,一个CountDownLatch变量就相当于golang中的sync.WaitGroup.wait就相当于其中的wait,而countDown则相当于被等待线程要调用的Done.