剖析muduo网络库核心代码

/*
muduo网络库给用户提供了两个主要的类
TcpServer:用于编写服务器程序
TcpClient:用于编写客户端程序

epoll+线程池
好处：能够把网络I/O的代码和业务代码区分开
                        用户的连接和断开，用户的可读写事件
*/
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <iostream>
#include <functional>
#include <string>
using namespace muduo::net;
using namespace muduo;
using namespace std;
/*基于muduo网络库开发服务器程序
1.组合TcpServer对象
2.创建EventLoop事件循环对象
3.明确TcpServer构造函数需要什么参数，输出ChatServer的构造函数
4.在当前服务器类的构造函数当中，注册处理连接的回调函数和处理读写事件的回调函数
5.设置合适的服务端线程数量，muduo库会自己分配I/O线程和worker线程
*/
class ChatServer
{
public:
    ChatServer(EventLoop* loop,//事件循环
            const InetAddress& listenAddr,//IP+PORT
            const string& nameArg)//服务器的名字
            :_server(loop,listenAddr,nameArg)
            ,_loop(loop)
            {
                //给服务器注册注册用户连接的创建和断开的回调
                _server.setConnectionCallback(std::bind(&ChatServer::onConnection,this,placeholders::_1));
                //给服务器注册用户读写事件回调
                _server.setMessageCallback(std::bind(&ChatServer::onMessage,this,placeholders::_1,_2,_3));
                //设置服务器端的线程数量   1个I/O线程 3个worker线程
                _server.setThreadNum(4);
            }
    //开启事件循环
    void start()
    {
        _server.start();
    }
private:
    //专门处理用户的连接创建和断开 
    void onConnection(const TcpConnectionPtr&conn)
    {
       
        if(conn->connected())
        {
            std::cout<<conn->peerAddress().toIpPort()<<"->"
            <<conn->localAddress().toIpPort()<<"state:online"<<std::endl;
        }
        else
        {
            std::cout<<conn->peerAddress().toIpPort()<<"->"
            <<conn->localAddress().toIpPort()<<"state:offline"<<std::endl;
            conn->shutdown();//close(fd);
        }
    }
    //专门处理用户的读写事件
    void onMessage(const TcpConnectionPtr&conn,//连接
                Buffer*buffer,//缓冲区
                Timestamp time)//接收到数据的时间信息
    {
        std::string buf=buffer->retrieveAllAsString();
        std::cout<<"recv data: "<<buf<<"time:"<<time.toString()<<std::endl;
        conn->send(buf);
    }
    muduo::net::TcpServer _server; //#1
    muduo::net::EventLoop *_loop;//#2 epoll
};
int main()
{
    EventLoop loop;
    InetAddress addr("127.0.0.1",6000);
    ChatServer server(&loop,addr,"ChatServer");
    server.start();//listenfd epoll_ctl=>epoll
    loop.loop();//epoll_wait以阻塞方式等待新用户的连接，已连接用户的读写事件等
    return 0;
}

#pragma once
/*
noncopyable被继承后，派生类对象可以正常的构造和析构，但是派生类对象无法进行拷贝构造和赋值操作
*/
class noncopyable
{
public:
    noncopyable(const noncopyable&) = delete;
    noncopyable& operator=(const noncopyable&)=delete;

protected:
    noncopyable()=default;
    ~noncopyable()=default;
};

#pragma once

#include <iostream>
#include <string>

//时间类
class Timestamp
{
public:
    Timestamp();
    //explicit让构造函数/转换函数只能“显式调用”，避免编译器偷偷帮你做类型转换，提升代码安全性。主要用于单个参数
    explicit Timestamp(int64_t microSecondsDinceEpoch);
    static Timestamp now();
    std::string toString() const;
private:
    int64_t microSecondsDinceEpoch_;
};

#include "Timestamp.h"

#include <time.h>
Timestamp::Timestamp():microSecondsDinceEpoch_(0){}
Timestamp::Timestamp(int64_t microSecondsDinceEpoch)    
    :microSecondsDinceEpoch_(microSecondsDinceEpoch)
    {}
Timestamp Timestamp::now()
{
    return Timestamp(time(NULL));
}
std::string Timestamp::toString() const
{
    char buf[128]={0};
    tm *tm_time=localtime(&microSecondsDinceEpoch_);
    //%02d按两位数输出整数，不足两位前面补 0。
    snprintf(buf,128,"%4d/%02d/%02d %02d:%02d:%02d",
        tm_time->tm_year+1900,
        tm_time->tm_mon+1,
        tm_time->tm_mday,
        tm_time->tm_hour,
        tm_time->tm_min,
        tm_time->tm_sec);
    return buf;
}
// #include <iostream>
// int main()
// {
//     std::cout<<Timestamp::now().toString()<<std::endl;
//     return 0;
// }

#pragma once
#include "noncopyable.h"

#include <string>

//LOG_INFO("%S %d",arg1,arg2) 宏定义里的 反斜杠 \ 是用来 换行但不结束宏定义 的。
#define LOG_INFO(LogmsgFormat,...)\
    do\
    {\
        Logger &logger=Logger::instance();\
        logger.setLogLevel(INFO);\
        char buf[1024]={0};\
        snprintf(buf,1024,LogmsgFormat,##__VA_ARGS__);\
        logger.log(buf);\
    }while(0)

#define LOG_ERROR(LogmsgFormat,...)\
    do\
    {\
        Logger &logger=Logger::instance();\
        logger.setLogLevel(ERROR);\
        char buf[1024]={0};\
        snprintf(buf,1024,LogmsgFormat,##__VA_ARGS__);\
        logger.log(buf);\
    }while(0)

#define LOG_FATAL(LogmsgFormat,...)\
    do\
    {\
        Logger &logger=Logger::instance();\
        logger.setLogLevel(FATAL);\
        char buf[1024]={0};\
        snprintf(buf,1024,LogmsgFormat,##__VA_ARGS__);\
        logger.log(buf);\
        exit(-1);\
    }while(0)

//如果定义了这个宏，就会输出DEBUG信息，没有就不输出
#ifdef MUDEBUG
#define LOG_DEBUG(LogmsgFormat,...)\
    do\
    {\
        Logger &logger=Logger::instance();\
        logger.setLogLevel(DEBUG);\
        char buf[1024]={0};\
        snprintf(buf,1024,LogmsgFormat,##__VA_ARGS__);\
        logger.log(buf);\
    }while(0)
#else
    #define LOG_DEBUG(LogmsgFormat,...)
#endif
//定义日志的级别 INFO ERROR FATAL DEBUG
enum LogLevel
{
    INFO,//错误信息
    ERROR,//错误信息
    FATAL,//core信息
    DEBUG,//调试信息
};
//输出一个日志类
class Logger:noncopyable
{
public:
    //获取日志唯一的实例对象
    static Logger& instance();
    //设置日志级别
    void setLogLevel(int level);
    //写日志
    void log(std::string msg);
private:
    int logLevel_;
    Logger(){}
};

#include "Logger.h"
#include "Timestamp.h"

#include <iostream>
//获取日志唯一的实例对象
Logger& Logger::instance()
{
    static Logger logger;
    return logger;
}
//设置日志级别
void Logger::setLogLevel(int level)
{
    logLevel_=level;
}
//写日志 [级别信息] time : msg
void Logger::log(std::string msg)
{
    switch (logLevel_)
    {
    case INFO:
        std::cout<<"[INFO]";
        break;
    case ERROR:
        std::cout<<"[ERROR]";
        break;
    case FATAL:
        std::cout<<"[FATAL]";
        break;
    case DEBUG:
        std::cout<<"[DEBUG]";
        break; 
    default:
        break;
    }
    
    //打印时间和msg
    std::cout << Timestamp::now().toString() << " : "<< msg <<std::endl;
}

#pragma once

#include <arpa/inet.h>
#include <netinet/in.h>
#include <string>
//封装socket地址类型
class InetAddress
{
public:
    explicit InetAddress(uint16_t port,std::string ip="127.0.0.1");
    explicit InetAddress(const sockaddr_in &addr):addr_(addr)
    {}
    std::string toIp() const;
    std::string toIpPort() const;
    uint16_t toPort() const;
    const sockaddr_in* getSockAddr() const {return &addr_;}
    void setSockAddr(const sockaddr_in &addr) {addr_=addr;}
private:
    sockaddr_in addr_;
};

#include "InetAddress.h"

#include <string.h>
InetAddress::InetAddress(uint16_t port, std::string ip)
{
    //保证 addr_ 结构体在使用前全为 0，避免未初始化内存带来的问题
    bzero(&addr_,sizeof(addr_));
    addr_.sin_family=AF_INET;
    addr_.sin_port=htons(port);
    addr_.sin_addr.s_addr=inet_addr(ip.c_str());
}
std::string InetAddress::toIp() const
{
    char buf[64]={0};
    inet_ntop(AF_INET,&addr_.sin_addr,buf,sizeof(buf));
    return buf;
}
std::string InetAddress::toIpPort() const
{
    char buf[64]={0};
    inet_ntop(AF_INET,&addr_.sin_addr,buf,sizeof(buf));
    size_t end=strlen(buf);
    uint16_t port=ntohs(addr_.sin_port);
    snprintf(buf+end,sizeof(buf)-end,":%u",port);
    return buf;
}
uint16_t InetAddress::toPort() const
{
    return ntohs(addr_.sin_port);
}
// #include <iostream>
// int main()
// {
//     InetAddress addr(8080);
//     std::cout<<addr.toIpPort()<<std::endl;
//     return 0;
// }

#pragma once

#include <iostream>
#include <string>

//时间类
class Timestamp
{
public:
    Timestamp();
    //explicit让构造函数/转换函数只能“显式调用”，避免编译器偷偷帮你做类型转换，提升代码安全性。主要用于单个参数
    explicit Timestamp(int64_t microSecondsDinceEpoch);
    static Timestamp now();
    std::string toString() const;
private:
    int64_t microSecondsDinceEpoch_;
};

#include "Timestamp.h"

#include <time.h>
Timestamp::Timestamp():microSecondsDinceEpoch_(0){}
Timestamp::Timestamp(int64_t microSecondsDinceEpoch)    
    :microSecondsDinceEpoch_(microSecondsDinceEpoch)
    {}
Timestamp Timestamp::now()
{
    return Timestamp(time(NULL));
}
std::string Timestamp::toString() const
{
    char buf[128]={0};
    tm *tm_time=localtime(&microSecondsDinceEpoch_);
    //%02d按两位数输出整数，不足两位前面补 0。
    snprintf(buf,128,"%4d/%02d/%02d %02d:%02d:%02d",
        tm_time->tm_year+1900,
        tm_time->tm_mon+1,
        tm_time->tm_mday,
        tm_time->tm_hour,
        tm_time->tm_min,
        tm_time->tm_sec);
    return buf;
}
// #include <iostream>
// int main()
// {
//     std::cout<<Timestamp::now().toString()<<std::endl;
//     return 0;
// }

#pragma once

#include "noncopyable.h"
#include "Timestamp.h"

#include <functional>
#include <memory>
/*
关于前置声明 class EventLoop;
什么时候需要 include？
    1.如果类成员是对象（非指针/引用），编译器需要知道对象的大小 → 必须 include。
    2.如果函数参数/返回值是对象（非指针/引用），也需要完整定义。→ 必须 include。
    3.如果只是声明指针、引用，或者在函数里只用到“声明”，那就只 forward declare。

前向声明 vs include 是 编译期依赖管理问题。
    你做成动态库时，头文件暴露给使用者。
    如果使用者只需要知道有 class EventLoop; 就能编译，那库的头文件就不用 include EventLoop.h。
    这样用户编译时依赖更少，只在链接时才需要 .so 提供实现。

能 forward declare 就 forward declare，只有在需要完整定义时才 include。
这样做可以减少编译依赖，加快编译速度，降低耦合。
*/
class EventLoop;
/*
Channel理解为通道，
封装了sockfd和其感兴趣的event,如EPOLLIN,EPOLLOUT事件，
还绑定了poller返回的具体事件
*/
class Channel : noncopyable
{
public:
    using EventCallback = std::function<void()>;
    using ReadEventCallback = std::function<void(Timestamp)>;
    Channel(EventLoop *loop, int fd);
    ~Channel();

    // fd得到Poller通知以后，处理事件
    void handlerEvent(Timestamp receiveTime);

    // 设置回调函数对象
    /*
    这里回调函数使用值传递+move。
    为什么不把参数设置为引用？首先如果使用lambda对象就不能把参数设置为普通引用。反而如果使用lambda，
    值传递+move这种模式效率更高
    */
    void setReadCallback(ReadEventCallback cb)
    {
        readCallback_ = std::move(cb);
    }
    void setWriteCallback(EventCallback cb)
    {
        writeCallback_ = std::move(cb);
    }
    void setCloseCallback(EventCallback cb)
    {
        closeCallback_ = std::move(cb);
    }
    void setErrorCallback(EventCallback cb)
    {
        errorCallback_ = std::move(cb);
    }

    // // channel的tie方法什么时候调用？TcpConnection => Channel
    void tie(const std::shared_ptr<void> &);

    int fd() const
    {
        return fd_;
    }
    int events() const { return events_; }
    void set_revents(int revt) { revents_ = revt; }

    // 设置fd相应的事件状态
    void enableReading()
    {
        events_ |= kReadEvent;
        update();
    }
    void disableReading()
    {
        events_ &= ~kReadEvent;
        update();
    }
    void enableWriting()
    {
        events_ |= kWriteEvent;
        update();
    }
    void disableWriting()
    {
        events_ &= ~kWriteEvent;
        update();
    }
    void disableAll()
    {
        events_ = kNoneEvent;
        update();
    }

    // 返回fd当前的事件状态
    bool isNoneEvent() const { return events_ == kNoneEvent; }
    bool isWriting() const { return events_ & kWriteEvent; }
    bool isReading() const { return events_ & kReadEvent; }

    int index() { return index_; }
    void set_index(int idx) { index_ = idx; }

    // one loop per thread
    EventLoop *ownerLoop() { return loop_; }

    void remove();

private:
    void update();
    void handleEventWithGuard(Timestamp receiveTime);

    static const int kNoneEvent;
    static const int kReadEvent;
    static const int kWriteEvent;

    EventLoop *loop_; // 事件循环
    const int fd_;    // fd,Poller监听的对象
    int events_;      // 注册fd感兴趣的事件
    int revents_;     // Poller返回的具体发生的事件
    int index_;

    std::weak_ptr<void> tie_;
    bool tied_;

    // 因为Channel通道里面能够获知fd最终发生的具体的事件的revents，所以它负责调用具体事件的回调操作
    ReadEventCallback readCallback_;
    EventCallback writeCallback_;
    EventCallback closeCallback_;
    EventCallback errorCallback_;
};

#include "Channel.h"
#include "EventLoop.h"
#include "Logger.h"

#include <sys/epoll.h>
const int Channel::kNoneEvent = 0;
const int Channel::kReadEvent = EPOLLIN | EPOLLPRI;
const int Channel::kWriteEvent = EPOLLOUT;

Channel::Channel(EventLoop *loop, int fd)
    : loop_(loop), fd_(fd), events_(0), revents_(0), index_(-1), tied_(false)
{
}
Channel::~Channel()
{
}

// 防止当channel被手动remove掉，channel还在执行回调操作
void Channel::tie(const std::shared_ptr<void> &obj)
{
    tie_ = obj;
    tied_ = true;
}

/*
当改变channel所表示fd的events事件后，update负责在Poller里面更改fd相应的事件epoll_ctl
EventLoop => ChannelList Poller
*/
void Channel::update()
{
    // 通过Channel所属的EventLoop,调用Poller的相应方法，注册fd的events事件
    loop_->updateChannel(this)
}
// 在channel所属的EventLoop中，把当前的Channel删除掉
void Channel::remove()
{
    loop_->removeChannel(this);
}

// fd得到Poller通知以后，处理事件
void Channel::handlerEvent(Timestamp receiveTime)
{
    if (tied_)
    {
        std::shared_ptr<void> guard = tie_.lock();
        if (guard)
        {
            handleEventWithGuard(receiveTime);
        }
    }
    else
    {
        handleEventWithGuard(receiveTime);
    }
}

// 根据Poller通知的channel发生的具体事件，由channel负责调用具体的回调操作
void Channel::handleEventWithGuard(Timestamp receiveTime)
{
    LOG_INFO("Channel handlerEvent revents:%d", revents_);
    // EPOLLHUP：对端关闭连接（挂起事件）
    // 判断“只有挂起且不可读”才触发关闭回调
    if ((revents_ & EPOLLHUP) && !(revents_ & EPOLLIN))
    {
        if (closeCallback_)
        {
            closeCallback_();
        }
    }

    if (revents_ & EPOLLERR)
    {
        if (errorCallback_)
        {
            errorCallback_();
        }
    }
    if (revents_ & (EPOLLIN | EPOLLPRI))
    {
        if (readCallback_)
        {
            readCallback_(receiveTime);
        }
    }
    if (revents_ & EPOLLOUT)
    {
        if (writeCallback_)
        {
            writeCallback_();
        }
    }
}

#pragma once

#include "noncopyable.h"
#include "Timestamp.h"

#include <vector>
#include <unordered_map>
class Channel;
class EventLoop;
// muduo库中多路分发器的核心IO复用模块
class Poller : noncopyable
{
public:
    using ChannelList = std::vector<Channel *>;
    Poller(EventLoop *loop);
    virtual ~Poller() = default;

    // 给所有IO复用保留统一的接口
    virtual Timestamp poll(int timeoutMs, ChannelList *activeChannels) = 0;
    virtual void updateChannel(Channel *channel) = 0;
    virtual void removeChannel(Channel *channel) = 0;

    // 判断参数channel是否在当前Poller当中
    bool hasChannel(Channel *channel) const;

    // EventLoop可以通过该接口获取默认的IO复用的具体实现
    // 但是这个不可以在Poller层实现，必须有具体的类才可以
    //所以我们新建DefaPoller.cc把实现写到这里，这样就避免include子类的问题
    //因为需要不依赖对象，所以用static
    static Poller *newDefaultPoller(EventLoop *loop);

protected:
    // map的key:sockfd value:sockfd所属的Channel通道类型
    using ChannelMap = std::unordered_map<int, Channel *>;
    ChannelMap channels_;

private:
    EventLoop *ownerLoop_; // 定义Poller所属的事件循环EventLoop
};

#include "Poller.h"
#include "Channel.h"
Poller::Poller(EventLoop *loop)
    : ownerLoop_(loop)
{
}
// 判断参数channel是否在当前Poller当中
bool Poller::hasChannel(Channel *channel) const
{
    auto it = channels_.find(channel->fd());
    return it != channels_.end() && it->second == channel;
}

#include "Poller.h"
#include "EPollPoller.h"

#include <stdlib.h>

// EventLoop可以通过该接口获取默认的IO复用的具体实现
// 但是这个不可以在Poller层实现，必须有具体的类才可以
// 所以我们新建DefaPoller.cc把实现写到这里，这样就避免include子类的问题
Poller *Poller::newDefaultPoller(EventLoop *loop)
{
    if (getenv("MUDUO_USE_POLL"))
    {
        return nullptr; // 生成poll的实例
    }
    else
    {
        return new EPollPoller(loop); // 生成epoll的实例
    }
}

#pragma once

#include <vector>
#include <sys/epoll.h>

#include "Poller.h"

class Channel;
/*
epoll的使用
epoll_create
epoll_ctl add/mod/del
epoll_wait
*/
class EPollPoller : public Poller
{
public:
    EPollPoller(EventLoop *loop);
    ~EPollPoller() override; // override让编译器检查这个是否在父类是虚函数

    // 重写基类的抽象方法
    Timestamp poll(int timeoutMs, ChannelList *activeChannels) override;
    void updateChannel(Channel *channel) override;
    void removeChannel(Channel *channel) override;

private:
    static const int kInitEventListSize = 16;

    // 填写活跃的连接
    void fillActiveChannels(int numEvents, ChannelList *activeChannels) const;
    // 更新channel通道
    void update(int operation, Channel *channel);

    using EventList = std::vector<epoll_event>;
    int epollfd_;
    EventList events_;
};

#include "EPollPoller.h"
#include "Logger.h"
#include "errno.h"
#include "Channel.h"

#include <cstring>
#include <unistd.h>
// Channel未添加到Poller中
const int kNew = -1; // Channel的成员index_=-1;
// Channel已添加到Poller中
const int kAdded = 1;
// Channel从Poller中删除
const int kDeleted = 2;
/*
epollfd_(epoll_create1(EPOLL_CLOEXEC)):
EPOLL_CLOEXEC = 创建 epoll fd 的时候就带上 close-on-exec 标志，
确保进程调用 exec() 时这个 fd 不会被子进程继承，避免 fd 泄露。
*/
EPollPoller::EPollPoller(EventLoop *loop)
    : Poller(loop), epollfd_(epoll_create1(EPOLL_CLOEXEC)), events_(kInitEventListSize)
{
    if (epollfd_ < 0)
    {
        LOG_FATAL("epoll_create error:%d \n", errno);
    }
}
EPollPoller::~EPollPoller()
{
    close(epollfd_);
}

// 重写基类的抽象方法
Timestamp EPollPoller::poll(int timeoutMs, ChannelList *activeChannels)
{
    // 实际上应该LOG_DEBUG输出日志更为合理
    LOG_INFO("func=%s -> fd total count:%lu\n", __FUNCTION__, channels_.size());
    /*
    &*events_.begin()为什么不用&events_[0]？
        1.&events_[0] 只有在 events_ 非空 时才合法；
        如果 events_ 的 size()==0，访问 events_[0] 就是 UB（未定义行为）。
        2.&*events_.begin() 即使 events_ 是空的，也不会立刻触发未定义行为（begin() == end()，
        解引用其实还是 UB，但很多实现里更可控，而且用法约定俗成）。
        3.C++11 以后更推荐用 events_.data()，它即使在空 vector 上返回的也是合法指针（但不能解引用）。
    但是如果在调用 epoll_wait 之前 给 vector 提前 resize 了，那么 &events_[0] 就一定安全了，就可以用&events_[0]
    */
    int numEvents = ::epoll_wait(epollfd_, &*events_.begin(), static_cast<int>(events_.size()), timeoutMs);
    int savedError = errno;
    Timestamp now(Timestamp::now());
    if (numEvents > 0)
    {
        LOG_INFO("%d events happened\n", numEvents);
        fillActiveChannels(numEvents, activeChannels);
        if (numEvents == events_.size())
        {
            events_.resize(events_.size() * 2);
        }
    }
    else if (numEvents == 0)
    {
        LOG_DEBUG("%s timeout!\n", __FUNCTION__);
    }
    else
    {
        if (savedError != EINTR)
        {
            errno = savedError;
            LOG_ERROR("EPOLLPoller::poll() err!");
        }
    }
    return now;
}
// channel update remove =>EventLoop updateChannel removeChannel
//=>Poller updateChannel removeChannel
/*
          EventLoop => poller.poll
    ChannelList     Poller
                    ChannelMap <fd,channel*> epollfd
*/
void EPollPoller::updateChannel(Channel *channel)
{
    const int index = channel->index();
    LOG_INFO("func=%s => fd=%d events=%d index=%d \n", __FUNCTION__, channel->fd(), channel->events(), index);
    if (index == kNew || index == kDeleted)
    {
        if (index == kNew)
        {
            int fd = channel->fd();
            channels_[fd] = channel;
        }
        channel->set_index(kAdded);
        update(EPOLL_CTL_ADD, channel);
    }
    else // channel已经在poller上注册过了
    {
        int fd = channel->fd();
        if (channel->isNoneEvent())
        {
            update(EPOLL_CTL_DEL, channel);
            channel->set_index(kDeleted);
        }
        else
        {
            update(EPOLL_CTL_MOD, channel);
        }
    }
}
// 从poller中删除channel
void EPollPoller::removeChannel(Channel *channel)
{
    int fd = channel->fd();
    channels_.erase(fd);
    LOG_INFO("func=%s => fd=%d \n", __FUNCTION__, channel->fd());
    int index = channel->index();
    if (index == kAdded)
    {
        update(EPOLL_CTL_DEL, channel);
    }
    channel->set_index(kNew);
}

// 填写活跃的连接
void EPollPoller::fillActiveChannels(int numEvents, ChannelList *activeChannels) const
{
    for (int i = 0; i < numEvents; ++i)
    {
        Channel *channel = static_cast<Channel *>(events_[i].data.ptr);
        channel->set_revents(events_[i].events);
        activeChannels->push_back(channel); // EventLoop就拿到了它的Poller给他返回的所有发生事件的channel列表
    }
}
// 更新channel通道
void EPollPoller::update(int operation, Channel *channel)
{
    epoll_event event;
    memset(&event, 0, sizeof(event));
    event.events = channel->events();
    event.data.ptr = channel;
    int fd = channel->fd();

    if (::epoll_ctl(epollfd_, operation, fd, &event) < 0)
    {
        LOG_ERROR("epoll_ctl del error:%d\n", errno);
    }
    else
    {
        LOG_FATAL("epoll_ctl del error:%d\n", errno);
    }
}

#pragma once

/*
为什么用namespace而不用类？
    1.类 (class)：表示有 状态、需要实例化的东西，比如 Channel、Poller、TcpConnection。
    2.命名空间 (namespace)：表示一组 无状态的工具函数/变量，比如 CurrentThread、std::chrono
    对于一些工作函数可以封装到域中。
*/
namespace CurrentThread
{
    /*
    为什么使用__thread?
        __thread 是 GCC/Clang 提供的 线程局部存储 (Thread Local Storage, TLS) 关键字。
        意思是：每个线程都会有一个独立的 t_cachedTid 变量，互不干扰。
        这样做的好处是：避免加锁开销，又能保证每个线程都能缓存自己的线程 ID。
    */
    /*
    为什么使用extern?
        1.extern表示声明，不分配内存，定义在别处（一般跨文件），不extern声明报错
        2.用于全局变量 / 跨文件共享时，头文件写 extern，cpp 文件写定义。
        3.函数声明默认就是 extern。
    */
    extern __thread int t_cachedTid;
    void cacheTid();
    inline int tid()
    {
        if(__builtin_expect(t_cachedTid==0,0))
        {
            cacheTid();
        }
        return t_cachedTid;
    }
}

#include "CurrentThread.h"

#include <unistd.h>
#include <sys/syscall.h>
namespace CurrentThread
{
    __thread int t_cachedTid;

    void cacheTid()
    {
        if(t_cachedTid==0)
        {
            //通过Linux系统调用，获取当前线程的tid值
            t_cachedTid=static_cast<pid_t>(::syscall(SYS_gettid));
        }
    }
}

#pragma once

#include <functional>
#include <vector>
#include <atomic>
#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()>;

    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 handlerRead();       // wake up
    void doPendingFunctors(); // 执行回调
    using ChannelList = std::vector<Channel *>;

    std::atomic_bool looping_;
    std::atomic_bool quit_; // 标识退出loop循环

    const pid_t threadId_; // 记录当前loop所在线程的id

    Timestamp pollReturnTime_; // poller返回发生事件的channels的时间点
    std::unique_ptr<Poller> poller_;

    /*
    主要作用:当mainLoop获取一个新用户的Channel,通过轮询选择一个subLoop,
    通过该成员唤醒subLoop处理channel
    */
    int wakeupFd_;
    std::unique_ptr<Channel> wakeupChannel_;

    ChannelList activeChannels_;

    std::atomic_bool callingPendingFunctors_; // 标识当前loop是否有需要执行的回调操作
    std::vector<Functor> PendingFunctors_;    // 存储loop需要执行的所有的回调操作
    std::mutex mutex_;                        // 互斥锁用来保护上面vector的线程安全
};

#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 EventLoop *t_loopThisThread = nullptr;

// 定义默认的Poller IO复用接口的超时时间
const int kPollTimeMs = 10000;

// 创建wakeupfd,用来notify唤醒subReactor处理新来的channel
/*
跨线程安全唤醒
eventfd返回值就是一个内核提供的原子计数器 fd，能被 epoll 监听，写就是“通知+1”，读就是“取值并清零”。
*/
int createEventfd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0)
    {
        LOG_FATAL("eventfd err:%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 create %p in thread %d\n", this, threadId_);
    if (t_loopThisThread)
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d \n", t_loopThisThread, threadId_);
    }
    else
    {
        t_loopThisThread = this;
    }

    // 设置wakeupfd的事件类型以及发生事件后的回调
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handlerRead, this));
    // 每一个eventloop都将监听wakeupchannel的EPOLLIN读事件
    wakeupChannel_->enableReading();
}
EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopThisThread = nullptr;
}
// 开启事件循环
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监听哪些channnel发生事件了，然后上报给EventLoop,通知channel处理相应的事件
            channel->handlerEvent(pollReturnTime_);
        }
        // 执行当前EventLoop事件循环需要处理的回调操作
        /*
        IO线程 mainLoop accept fd<=channel subLoop
        mainLoop 实现注册一个回调cb(需要subLoop执行) wakeup subLoop
        执行下面的方法，执行之前mainloop注册的cb
        */
        doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping", this);
    looping_ = false;
}
// 退出事件循环 1.loop在自己的线程调用quit 2.在非loop的线程中，调用loop的quit
void EventLoop::quit()
{
    quit_ = true;

    // 如果在其他线程中调用quit 在一个subloop(worker)中调用了mainLoop(IO)的quit，
    // 由于不知道状态，所以要wakeup
    if (!isInLoopThread())
    {
        wakeup();
    }
}

// 在当前loop执行cb
void EventLoop::runInLoop(Functor cb)
{
    // 在当前的loop线程执行cb
    if (isInLoopThread())
    {
        cb();
    }
    else // 在非当前loop中执行cb,就需要唤醒loop所在线程，执行cb
    {
        queueInLoop(cb);
    }
}
// 把cb放入队列中，唤醒loop所在的线程，执行cb
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lokc(mutex_);
        PendingFunctors_.emplace_back(std::move(cb));
    }
    // 唤醒相应的，需要执行上面回调操作的loop的线程了
    // || callingPendingFunctors_的意思：如果当前loop正在执行回调，但是loop又有了新的回调，所以需要唤醒，不能再阻塞
    if (!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup(); // 唤醒loop所在线程
    }
}

void EventLoop::handlerRead() // wake up
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::handleRead() reads %ld bytes instead of 8", n);
    }
}

// 用来唤醒loop所在的线程   向wakeupfd_写一个数据
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);
    }
}

// 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()
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(PendingFunctors_);
    }
    for (const Functor &functor : functors)
    {
        functor();//执行当前loop需要执行的回调操作
    }
    callingPendingFunctors_ = false;
}

#pragma once

#include "noncopyable.h"

#include <functional>
#include <thread>
#include <memory>
#include <unistd.h>
#include <string>
#include <atomic>
class Thread : noncopyable
{
public:
    using ThreadFunc = std::function<void()>;

    explicit Thread(ThreadFunc, const std::string &name = std::string());
    ~Thread();

    void start();
    void join();

    bool started() const { return started_; }
    pid_t tid() const { return tid_; }
    const std::string &name() const { return name_; }

    static int numCreated() { return numCreated_; }

private:
    void setDefaultName();
    bool started_;
    bool joined_;
    std::shared_ptr<std::thread> thread_;
    pid_t tid_;
    ThreadFunc func_;
    std::string name_;
    static std::atomic_int numCreated_;
};

#include "Thread.h"
#include "CurrentThread.h"

#include <semaphore.h>
std::atomic_int Thread::numCreated_(0);
Thread::Thread(ThreadFunc func, const std::string &name)
:started_(false)
,joined_(false)
,tid_(0)
,func_(std::move(func))
,name_(name)
{
    setDefaultName();
}
Thread::~Thread()
{
    if(started_&&!joined_)
    {
        thread_->detach();//设置分离
    }
}

void Thread::start()
{
    started_=true;
    sem_t sem;
    sem_init(&sem,false,0);
    //开启线程
    thread_=std::shared_ptr<std::thread>(new std::thread([&](){
        //获取线程的tid值
        tid_=CurrentThread::tid();
        sem_post(&sem);
        func_();//开启一个线程，专门执行该线程函数
    }));

    //这里必须等待获取上面新创建的线程的tid值
    sem_wait(&sem);
}
void Thread::join()
{
    joined_=true;
    thread_->join();
}
void Thread::setDefaultName()
{
    int num=++numCreated_;
    if(name_.empty())
    {
        char buf[32]={0};
        snprintf(buf,sizeof(buf),"Thread%d",num);
        name_=buf;
    }
}

#pragma once

#include <functional>
#include <mutex>
#include <condition_variable>
#include <string>

#include "noncopyable.h"
#include "Thread.h"
class EventLoop;
class EventLoopThread : noncopyable
{
public:
    using ThreadInitCallback = std::function<void(EventLoop *)>;
    EventLoopThread(const ThreadInitCallback &cb = ThreadInitCallback(), const std::string &name = std::string());
    ~EventLoopThread();

    EventLoop *startLoop();

private:
    void threadFunc();
    EventLoop *loop_;
    bool exiting_;
    Thread thread_;
    std::mutex mutex_;
    std::condition_variable cond_;
    ThreadInitCallback callback_;
};

#include "EventLoopThread.h"
#include "EventLoop.h"

#include <memory>
EventLoopThread::EventLoopThread(const ThreadInitCallback &cb, const std::string &name)
    : loop_(nullptr), exiting_(false), thread_(std::bind(&EventLoopThread::threadFunc, this), name), mutex_(), cond_(), callback_(cb)
{
}
EventLoopThread::~EventLoopThread()
{
    exiting_ = true;
    if (loop_ != nullptr)
    {
        loop_->quit();
        thread_.join();
    }
}

EventLoop *EventLoopThread::startLoop()
{
    thread_.start(); // 启动底层的新线程

    EventLoop *loop = nullptr;
    {
        std::unique_lock<std::mutex> lock(mutex_);
        while (loop_ == nullptr)
        {
            cond_.wait(lock);
        }
        loop = loop_;
    }
    return loop;
}

// 下面这个方法，是在单独的新线程里面运行的
void EventLoopThread::threadFunc()
{
    // 创建一个独立的eventloop,和上面的线程是一一对应的，one loop per thread
    EventLoop loop;

    if (callback_)
    {
        callback_(&loop);
    }

    {
        std::unique_lock<std::mutex> lock(mutex_);
        loop_ = &loop;
        cond_.notify_one();
    }

    loop.loop(); // EventLoop loop => Poller.poll
    std::unique_lock<std::mutex> lock(mutex_);
    loop_ = nullptr;
}

#pragma once
#include <functional>
#include <string>
#include <vector>
#include <memory>

#include "noncopyable.h"

class EventLoop;
class EventLoopThread;
class EventLoopThreadPool : noncopyable
{
public:
    using ThreadInitCallback = std::function<void(EventLoop *)>;
    EventLoopThreadPool(EventLoop *baseLoop, const std::string &nameArg);
    ~EventLoopThreadPool();
    void setThreadNum(int numThreads) { numThreads_ = numThreads; }
    void start(const ThreadInitCallback &cb = ThreadInitCallback());

    // 如果工作在多线程，baseLoop_默认以轮询的方式分配channel给subloop
    EventLoop *getNextLoop();

    std::vector<EventLoop*> getAllLoops();

    bool started() const {return started_;}

    const std::string& name()const{return name_;}

private:
    EventLoop *baseLoop_;
    std::string name_;
    bool started_;
    int numThreads_;
    int next_;
    std::vector<std::unique_ptr<EventLoopThread>> threads_;
    std::vector<EventLoop *> loops_;
};