muduo源码分析——线程同步中门栓类的设计与实现

编程入门行业动态更新时间:2024-10-04 23:29:50

muduo源码分析——<a href=https://www.elefans.com/category/jswz/34/1771240.html style= 线程同步中门栓类的设计与实现"/>

muduo源码分析——线程同步中门栓类的设计与实现

0 门栓类

最近在看muduo源码时，遇到门栓类这个概念，在此做一下剖析和整理。
门栓类，顾名思义，其作用类似于一个门控开关，用于控制和协调主线程与各子线程之间的启动顺序。
门栓类主要有以下两种作用方式：

用于所有子线程等待主线程发起“起跑”

–>主线程发起“起跑”命令后，子线程才可以执行；
–>主线程要通知所有的子线程；
用于主线程等待子线程初始化完毕才可以工作

一个门栓类的实现是建立在互斥信号量和条件信号量的基础上的。

1 互斥信号量实现

class CAPABILITY("mutex") MutexLock : noncopyable
{public:MutexLock(): holder_(0)// holder_初始化为0，表示该锁没有被任何进程拥有{MCHECK(pthread_mutex_init(&mutex_, NULL));}~MutexLock(){// 断言当前锁没有被任何进程拥有，才可以销毁该锁assert(holder_ == 0);// 销毁锁MCHECK(pthread_mutex_destroy(&mutex_));}// 是否当前进程拥有该锁// must be called when locked, i.e. for assertionbool isLockedByThisThread() const{// 判断holder_是否等于当前进程的pidreturn holder_ == CurrentThread::tid();}// 断言当前进程拥有该锁void assertLocked() const ASSERT_CAPABILITY(this){assert(isLockedByThisThread());}// internal usage// 加锁void lock() ACQUIRE(){MCHECK(pthread_mutex_lock(&mutex_));// 调用pthread_mutex_lock()实现// 将当前进程的id赋值给holder_成员变量 holder_ = CurrentThread::tid();assignHolder();}// 解锁void unlock() RELEASE(){// holder_重置为0 holder_ = 0;unassignHolder();MCHECK(pthread_mutex_unlock(&mutex_));}// 获取mutex对象pthread_mutex_t* getPthreadMutex() /* non-const */{// 返回mutex_成员变量return &mutex_;}private:friend class Condition;class UnassignGuard : noncopyable{public:explicit UnassignGuard(MutexLock& owner): owner_(owner){owner_.unassignHolder();}~UnassignGuard(){owner_.assignHolder();}private:MutexLock& owner_;};void unassignHolder(){holder_ = 0;}void assignHolder(){holder_ = CurrentThread::tid();}pthread_mutex_t mutex_;pid_t holder_;// 当前拥有该互斥锁的线程id tid?
};// Use as a stack variable, eg.
// int Foo::size() const
// {
//   MutexLockGuard lock(mutex_);
//   return data_.size();
// }
// 类MutexLockGuard
// 使用RAII技法封装 资源获取即初始化
// 使用类来管理资源 利用类的构造、析构函数以及对象的生命周期 作用域// 为什么使用MutexLockGuard类?--避免忘记解锁
// 只使用MutexLock类存在加锁而不能及时解锁的问题 如下：
// MutexLcok mutex;
// void f()
// {
//     mutex.lock();
//     ...
//     if (条件)
//     {
//       mutex.unlock();// 提前解锁
//       return;
//     }
//     mutex.unlock();
// }// 使用了MutexLockGuard之后
// MutexLcok mutex;
// void f()
// {
//     MutexLockGuard lock(mutex);// 该对象的作用域为f()函数 函数体
//     ...
//     if (条件)
//     {
//       return;// 函数return后，MutexLockGuard类的对象lock也会失效--离开作用域，
//       自动调用析构函数(mutex_.unlock())将mutex解锁
//     }
// }
class SCOPED_CAPABILITY MutexLockGuard : noncopyable
{public:// 获取资源// mutex对象的生成和销毁不归MutexLockGuard类管理// MutexLock与MutexLockGuard类的关系是关联关系explicit MutexLockGuard(MutexLock& mutex) ACQUIRE(mutex): mutex_(mutex){// 调用lockmutex_.lock();}// 释放资源~MutexLockGuard() RELEASE(){mutex_.unlock();}private:MutexLock& mutex_;// 注意是mutex&引用 mutex对象的生成和销毁不归MutexLockGuard类管理
};

2 条件信号量实现

Condition.h

class Condition : noncopyable
{public:explicit Condition(MutexLock& mutex)// 不拥有 也不负责管理mutex变量的生存期: mutex_(mutex){// 创建条件变量pcond_MCHECK(pthread_cond_init(&pcond_, NULL));}~Condition(){// 销毁条件变量MCHECK(pthread_cond_destroy(&pcond_));}// wait函数void wait(){MutexLock::UnassignGuard ug(mutex_);// 调用pthread_cond_wait()函数MCHECK(pthread_cond_wait(&pcond_, mutex_.getPthreadMutex()));}// returns true if time out, false otherwise.bool waitForSeconds(double seconds);// 即signal()函数void notify(){// 调用pthread_cond_signal()函数MCHECK(pthread_cond_signal(&pcond_));}void notifyAll(){MCHECK(pthread_cond_broadcast(&pcond_));}private:MutexLock& mutex_;// MutexLock mutexLockpthread_cond_t pcond_;// 条件变量
};

Condition

// returns true if time out, false otherwise.
bool Condition::waitForSeconds(double seconds)
{struct timespec abstime;// FIXME: use CLOCK_MONOTONIC or CLOCK_MONOTONIC_RAW to prevent time rewind.clock_gettime(CLOCK_REALTIME, &abstime);const int64_t kNanoSecondsPerSecond = 1000000000;int64_t nanoseconds = static_cast<int64_t>(seconds * kNanoSecondsPerSecond);abstime.tv_sec += static_cast<time_t>((abstime.tv_nsec + nanoseconds) / kNanoSecondsPerSecond);abstime.tv_nsec = static_cast<long>((abstime.tv_nsec + nanoseconds) % kNanoSecondsPerSecond);MutexLock::UnassignGuard ug(mutex_);// 调用pthread_cond_timedwait()return ETIMEDOUT == pthread_cond_timedwait(&pcond_, mutex_.getPthreadMutex(), &abstime);
}

3 门栓类实现

CountDownLatch.h

class CountDownLatch : noncopyable
{public:explicit CountDownLatch(int count);// ???void wait();// 计数器减一-->计数器减为0之后便向所有的线程发起通知void countDown();// 获取当前计数器的值int getCount() const;private:mutable MutexLock mutex_;// mutable:可以在const修饰该变量时，改变该变量的值 参考CountDownLatch::getCount()Condition condition_ GUARDED_BY(mutex_);int count_ GUARDED_BY(mutex_);// 计数器
};

CountDownLatch

CountDownLatch::CountDownLatch(int count): mutex_(),// 创造一个mutex_对象condition_(mutex_),// 将mutex_对象传到condition_里面count_(count)
{
}void CountDownLatch::wait()
{MutexLockGuard lock(mutex_);// 保护count_值 可能多个线程都会访问count_变量while (count_ > 0){condition_.wait();}// 离开CountDownLatch::wait()函数的作用域后，lock会自动解锁
}void CountDownLatch::countDown()
{MutexLockGuard lock(mutex_);// 保护count值--count_;// count减1if (count_ == 0){condition_.notifyAll();// 通知所有的等待线程}
}int CountDownLatch::getCount() const
{// mutex_为mutable变量 可以修改MutexLockGuard lock(mutex_);// 保护count值return count_;// 返回count值
}

4 测试代码

测试代码一：主线程唤醒子线程
CountDownLatch_test1

#include <muduo/base/CountDownLatch.h>
#include <muduo/base/Thread.h>#include <boost/bind.hpp>
#include <boost/ptr_container/ptr_vector.hpp>
#include <string>
#include <stdio.h>using namespace muduo;class Test
{public:Test(int numThreads): latch_(1),threads_(numThreads){// 创建三个线程for (int i = 0; i < numThreads; ++i){char name[32];snprintf(name, sizeof name, "work thread %d", i);threads_.push_back(new muduo::Thread(boost::bind(&Test::threadFunc, this), muduo::string(name)));}// 启动各子线程 从begin到end 执行Thread::start函数for_each(threads_.begin(), threads_.end(), boost::bind(&Thread::start, _1));}// 发号施令void run(){// count_减1 减少至0时 就开始latch_.countDown();}void joinAll(){for_each(threads_.begin(), threads_.end(), boost::bind(&Thread::join, _1));}private:void threadFunc(){// 等待主线程发号施令latch_.wait();printf("tid=%d, %s started\n",CurrentThread::tid(),CurrentThread::name());printf("tid=%d, %s stopped\n",CurrentThread::tid(),CurrentThread::name());}CountDownLatch latch_;boost::ptr_vector<Thread> threads_;
};int main()
{printf("pid=%d, tid=%d\n", ::getpid(), CurrentThread::tid());Test t(3);sleep(3);printf("pid=%d, tid=%d %s running ...\n", ::getpid(), CurrentThread::tid(), CurrentThread::name());// 发号施令 主线程发号施令后各子线程才能执行t.run();t.joinAll();// 打印已启动的线程个数printf("number of created threads %d\n", Thread::numCreated());
}

测试代码二：各子线程唤醒主线程
CountDownLatch_test2

#include <muduo/base/CountDownLatch.h>
#include <muduo/base/Thread.h>#include <boost/bind.hpp>
#include <boost/ptr_container/ptr_vector.hpp>
#include <string>
#include <stdio.h>using namespace muduo;class Test
{public:Test(int numThreads): latch_(numThreads),threads_(numThreads){for (int i = 0; i < numThreads; ++i){char name[32];snprintf(name, sizeof name, "work thread %d", i);threads_.push_back(new muduo::Thread(boost::bind(&Test::threadFunc, this), muduo::string(name)));}for_each(threads_.begin(), threads_.end(), boost::bind(&muduo::Thread::start, _1));}void wait(){latch_.wait();// latch的count计数器初值为numThreads}void joinAll(){for_each(threads_.begin(), threads_.end(), boost::bind(&Thread::join, _1));}private:void threadFunc(){sleep(3);// 子线程负责countDown--将count_计数器减1 count_--latch_.countDown();// latch的count计数器减1printf("tid=%d, %s started\n",CurrentThread::tid(),CurrentThread::name());sleep(3);printf("tid=%d, %s stopped\n",CurrentThread::tid(),CurrentThread::name());}CountDownLatch latch_;boost::ptr_vector<Thread> threads_;
};int main()
{printf("pid=%d, tid=%d\n", ::getpid(), CurrentThread::tid());Test t(3);// 各子线程全部启动后主线程才能执行t.wait();printf("pid=%d, tid=%d %s running ...\n", ::getpid(), CurrentThread::tid(), CurrentThread::name());t.joinAll();printf("number of created threads %d\n", Thread::numCreated());
}