481-手写C++muduo库(EventLoop事件循环)

清疚 2022-09-08 11:50 332阅读 0赞

**EventLoop相当于reactor模型的reactor反应堆的角色**  
**poller和epollpoller相当于是多路分发器的角色，掌控epoll的所有操作**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_35_color_FFFFFF_t_70_g_se_x_16]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_40_color_FFFFFF_t_70_g_se_x_16]  
**最起码得有一个线程来支撑。**  
如果我们设置了setthreadnumber的多线程的反应堆类型，mainreactor就是处理新用户的连接，accept，拿到新用户通信的fd，把fd感兴趣的事件打包成channel，然后唤醒某个工作线程workreactor（subreactor）（轮询的方式），代表eventloop。每个工作线程subreactor都监听一组channel，每一组channel都得在自己的eventloop去执行。

**mainreactor和subreactor**  
如果没有事件发生的话，他们loop所在的线程都是阻塞的。  
假如现在mainreactor监听到一个新用户的连接，得到了表示新用户连接的fd，以及感兴趣的事件的channel，它把这个channel怎么扔给subreactor呢？  
如何叫醒？？？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_28_color_FFFFFF_t_70_g_se_x_16]  
**统一事件源的原理。**  
**在libevent库中，采用的是socketpair**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_54_color_FFFFFF_t_70_g_se_x_16]  
创建了一个本地socket的数组，这2个socket都是可读可写的。

**在muduo库中，采用的是linux提供的eventfd**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_57_color_FFFFFF_t_70_g_se_x_16]  
**这就是线程间的通信机制**。这个线程可以notify通知其他线程起来做事情。eventfd是系统API，在内核可以直接notify应用空间的应用程序，让它的线程起来做事情，效率高。

## EventLoop.h ##

#pragma once
    
    #include <functional>
    #include <vector>
    #include <atomic>//C++ 11
    #include <memory>
    #include <mutex>
    
    #include "noncopyable.h"
    #include "Timestamp.h"
    #include "CurrentThread.h"
    
    class Channel;
    class Poller;
    
    //时间循环类  主要包含了两个大模块 Channel   Poller（epoll的抽象）
    class EventLoop : noncopyable
    {
        
    public:
        using Functor = std::function<void()>;//定义一个回调的类型 
        //using代替typedef，进行类型的重命名 
    
        EventLoop();
        ~EventLoop();
    
        //开启事件循环
        void loop();
        //退出事件循环
        void quit();
    
        Timestamp pollReturnTime() const {
         return pollReturnTime_; }
        
        //在当前loop中执行cb
        void runInLoop(Functor cb);
        //把cb放入队列中，唤醒loop所在的线程，执行cb
        void queueInLoop(Functor cb);
    
        //用来唤醒loop所在的线程的
        void wakeup();
    
        //EventLoop的方法 =》 Poller的方法
        void updateChannel(Channel *channel);
        void removeChannel(Channel *channel);
        bool hasChannel(Channel *channel);
    
        //判断EventLoop对象是否在自己的线程里面
        bool isInLoopThread() const {
         return threadId_ ==  CurrentThread::tid(); }
    private:
        void handleRead();//wake up
        void doPendingFunctors();//执行回调
    
        using ChannelList = std::vector<Channel*>;
    
        std::atomic_bool looping_;//原子操作，通过CAS实现的
        std::atomic_bool quit_;//标识退出loop循环
        
        const pid_t threadId_;//记录当前loop所在线程的id
    
        Timestamp pollReturnTime_;//poller返回发生事件的channels的时间点
        std::unique_ptr<Poller> poller_;//eventloop所管理的poller 
    
        int wakeupFd_;//linux内核的eventfd创建出来的 
    	//主要作用，当mainLoop获取一个新用户的channel，通过轮询算法选择一个subloop，通过该成员唤醒subloop处理channel
    	std::unique_ptr<Channel> wakeupChannel_;//包括wakeupFd和感兴趣的事件 
    
        ChannelList activeChannels_;//eventloop所管理的channel 
    
        std::atomic_bool callingPendingFunctors_;//标识当前loop是否有需要执行的回调操作
        std::vector<Functor> pendingFunctors_;//存储loop需要执行的所有的回调操作
        std::mutex mutex_;//互斥锁，用来保护上面vector容器的线程安全操作
    };

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16]  
写执行回调的时候我们看到，它首先定义了一个局部的桩（vector），然后把1个变量置为true。然后加了1个锁，相当于把pendingFunctors\_置空，把放在里面的所有待执行的回调全部放到局部的vector对象里面，好处是：因为这个函数要一直不断从pendingFunctors\_拿1个回调执行拿1个回调执行，然后删掉，在这个时候，如果有很多回调，这里就一直锁着，导致mainloop阻塞住向你下发channel的操作上。造成服务器的时延变长。

![在这里插入图片描述][13bc480acf6f43f09c9a8fb7b5597262.png]

## EventLoop.cc ##

#include "EventLoop.h"
    #include "Logger.h"
    #include "Poller.h"
    #include "Channel.h"
    
    #include <sys/eventfd.h>
    #include <unistd.h>
    #include <fcntl.h>
    #include <errno.h>
    #include <memory>
    
    //防止一个线程创建多个EventLoop   __thread：thread_local
    //当一个eventloop创建起来它就指向那个对象，在一个线程里再去创建一个对象，由于这个指针为空，就不创建 
    __thread EventLoop *t_loopInThisThread = nullptr;
    
    //定义默认的Poller IO复用接口的超时时间
    const int kPollTimeMs = 10000;//10秒钟 
    
    //创建wakeupfd，用来notify唤醒subReactor处理新来的channel
    int createEventfd()
    {
        
        int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        if (evtfd < 0)
        {
        
            LOG_FATAL("eventfd error:%d \n", errno);
        }
        return evtfd;
    }
    
    EventLoop::EventLoop()//构造函数 
        : looping_(false)
        , quit_(false)
        , callingPendingFunctors_(false)
        , threadId_(CurrentThread::tid())
        , poller_(Poller::newDefaultPoller(this))
        , wakeupFd_(createEventfd())
        , wakeupChannel_(new Channel(this, wakeupFd_))
    {
        
        LOG_DEBUG("EventLoop created %p in thread %d \n", this, threadId_);
        if (t_loopInThisThread)//这个线程已经有loop了，就不创建了 
        {
        
            LOG_FATAL("Another EventLoop %p exists in this thread %d \n", t_loopInThisThread, threadId_);
        }
        else//这个线程还没有loop，创建 
        {
        
            t_loopInThisThread = this;
        }
    
        //设置wakeupfd的事件类型以及发生事件后的回调操作
        wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
        //每一个eventloop都将监听wakeupchannel的EPOLLIN读事件了
        wakeupChannel_->enableReading();
    }
    
    EventLoop::~EventLoop()//析构函数 
    {
        
        wakeupChannel_->disableAll();
        wakeupChannel_->remove();
        ::close(wakeupFd_);
        t_loopInThisThread = nullptr;
    }
    
    //开启事件循环 驱动底层的poller执行poll 
    void EventLoop::loop()
    {
        
        looping_ = true;
        quit_ = false;
    
        LOG_INFO("EventLoop %p start looping \n", this);
    
        while(!quit_)
        {
        
            activeChannels_.clear();
            //监听两类fd   一种是client的fd，一种wakeupfd
            pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
            for (Channel *channel : activeChannels_)
            {
        
                //Poller监听哪些channel发生事件了，然后上报给EventLoop，通知channel处理相应的事件
                channel->handleEvent(pollReturnTime_);//事先已经绑定好 
            }
            //执行当前EventLoop事件循环需要处理的回调操作
            /**
             * IO线程 mainLoop：accept  channel打包fd ---》 subloop 1个mainloop 3个subloop 
             * mainLoop 事先注册一个回调cb（需要subloop来执行）  mainloop wakeup subloop后，
    		 执行下面的方法，执行之前mainloop注册的cb操作
             */ 
            doPendingFunctors();//mainloop注册回调给subloop。 
        }
    
        LOG_INFO("EventLoop %p stop looping. \n", this);
        looping_ = false;
    }
    
    //退出事件循环  1.loop在自己的线程中调用quit  2.在非loop的线程中，调用loop的quit
    /**
     *              mainLoop
     * 
     *              通过wakeupfd                     no ==================== 生产者-消费者的线程安全的队列
                                                    mainloop生产 subloop消费  逻辑好处理 但是muduo库没有这个 是通过wakefd通信 
                                                    线程间直接notify唤醒 
     * 
     *  subLoop1     subLoop2     subLoop3    
     */ 
    void EventLoop::quit()
    {
        
        quit_ = true;
    
        //如果是在其它线程中，调用的quit 比如说：在一个subloop(woker)中，调用了mainLoop(IO)的quit
        //调用mainloop的quit，这时候，应该给它唤醒 ，它就从poll返回回来，
    	//当前线程在work线程  轮询的算法 派发channel 
        if (!isInLoopThread())  
        {
        
            wakeup();//因为不知道主线程是什么情况，需要唤醒一下 
        }
    }
    
    //之间在当前loop中执行cb（callback） 
    void EventLoop::runInLoop(Functor cb)
    {
        
        if (isInLoopThread())//在当前的loop线程中，执行cb
        {
        
            cb();
        }
        else//在非当前loop线程中执行cb , 就需要唤醒loop所在线程，执行cb
        {
        
            queueInLoop(cb);
        }
    }
    
    //不在当前的loop中执行，把cb放入队列中，唤醒loop所在的线程，执行cb
    //一个loop运行在自己的线程里。比如在subloop2调用subloop3的 runInLoop
    void EventLoop::queueInLoop(Functor cb)
    {
        
        {
        
            std::unique_lock<std::mutex> lock(mutex_);//因为有并发的访问 
            pendingFunctors_.emplace_back(cb);//直接构造cb放到vector里面 
        }
    
        //唤醒相应的，需要执行上面回调操作的loop的线程了
        // || callingPendingFunctors_的意思是：当前loop正在执行回调，但是loop又有了新的回调
        if (!isInLoopThread() || callingPendingFunctors_) 
        {
        
            wakeup();//唤醒loop所在线程，继续执行回调 
        }
    }
    
    void EventLoop::handleRead()//就是读，写啥读啥无所谓，就是为了唤醒loop线程执行回调 
    {
        
      uint64_t one = 1;
      ssize_t n = read(wakeupFd_, &one, sizeof one);
      if (n != sizeof one)
      {
        
        LOG_ERROR("EventLoop::handleRead() reads %lu bytes instead of 8", n);
      }
    }
    
    //用来唤醒loop所在的线程的  向wakeupfd_写一个数据，wakeupChannel就发生读事件，当前loop线程就会被唤醒
    void EventLoop::wakeup()
    {
        
        uint64_t one = 1;
        ssize_t n = write(wakeupFd_, &one, sizeof one);
        if (n != sizeof one)//子线程无法被唤醒 
        {
        
            LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8 \n", n);
        }
    }
    
    //EventLoop的方法 =》 Poller的方法
    void EventLoop::updateChannel(Channel *channel)
    {
        
        poller_->updateChannel(channel);
    }
    
    void EventLoop::removeChannel(Channel *channel)
    {
        
        poller_->removeChannel(channel);
    }
    
    bool EventLoop::hasChannel(Channel *channel)
    {
        
        return poller_->hasChannel(channel);
    }
    
    void EventLoop::doPendingFunctors()//执行回调 在loop中调用的方法 
    {
        
        std::vector<Functor> functors;
        callingPendingFunctors_ = true;
    
        {
        
            std::unique_lock<std::mutex> lock(mutex_);
            functors.swap(pendingFunctors_);//资源交换，把pendingFunctors_ 置为空
    		//不需要pendingFunctors_了  不妨碍 mainloop向 pendingFunctors_写回调操作cb 
        }
    
        for (const Functor &functor : functors)
        {
        
            functor();//执行当前loop需要执行的回调操作
        }
    
        callingPendingFunctors_ = false;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_35_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/632b0f0aad7d42bfa0420850adc7b5dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_40_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/ce201b77aeb1426da5cec1ca1b9da711.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_28_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/a667fbbbb7224a1eb3cfea78d56f5c69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_54_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/a542fd3e1f1049f9bbf4e5ebe8a957c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBALeael-azveWuhw_size_57_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/cfa71fe451f341049e8fa3771f737f27.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/f2ee2c604717429aa46d0c1ea5c520a1.png
[13bc480acf6f43f09c9a8fb7b5597262.png]: /images/20220829/46f19cbf01da4508b08730a72850261a.png