epoll实现多路转接¶

class EpollServer
{
public:
    EpollServer(uint16_t port = default_port)
        : bs_(std::make_shared<TcpSocket>()), isRunning_(false), efd_(-1)
    {
        bs_->initSocket();
        // 创建epoll模型
        efd_ = epoll_create(256);
    }

private:
    // ...
    int efd_;
};

EpollServer(uint16_t port = default_port)
    : bs_(std::make_shared<TcpSocket>(port)), isRunning_(false), efd_(-1)
{
    // ...
    // 关心listen_socketfd
    struct epoll_event ee;
    // 关心读事件
    ee.events = EPOLLIN;
    ee.data.fd = bs_->getListenSocketFd();
    int ret = epoll_ctl(efd_, EPOLL_CTL_ADD, bs_->getListenSocketFd(), &ee);
}

void startServer()
{
    isRunning_ = true;
    while (isRunning_)
    {
        // 不再需要在循环中多次设置监听套接字到读文件描述符集中
        // 等待1秒
        const int timeout = 1000;
        // 只关心读事件的文件描述符集
        int n = epoll_wait(efd_, fd_array_->data(), g_fd_array_num, timeout);
        if (n > 0)
        {
            // 监听套接字文件描述符准备完毕
            handler(n);
        }
        else if (n == 0)
        {
            LOG(LogLevel::INFO) << "没有客户端连接";
            sleep(1);
        }
    }
    isRunning_ = false;
}

void handler(int n)
{
    for (int i = 0; i < n; i++)
    {
        // 保证一定是就绪事件
        // 判断当前是否是监听套接字
        if((*fd_array_)[i].data.fd == bs_->getListenSocketFd())
        {
            // 获取链接
            SockAddrIn client;
            int ac_socketfd = bs_->toAccept(std::addressof(client));

            if(ac_socketfd > 0)
            {
                // 添加到epoll模型中
                struct epoll_event ev;
                ev.events = EPOLLIN;
                ev.data.fd = ac_socketfd;
                int ret = epoll_ctl(efd_, EPOLL_CTL_ADD, ac_socketfd, &ev);
                if (ret < 0)
                {
                    LOG(LogLevel::WARNING) << "添加事件失败";
                    exit(static_cast<int>(ErrorNumber::Epoll_Ctl_Fail));
                }
            }
        }
        else 
        {
            // 不是监听套接字
            std::string buffer;
            ssize_t ret = bs_->recvData(buffer, (*fd_array_)[i].data.fd);
            if(ret > 0)
            {
                LOG(LogLevel::INFO) << "客户端发送：" << buffer;
            }
            else if(ret == 0)
            {
                // 恢复操作
                // 从红黑树中移除指定文件描述符，此时不再需要关心事件
                int ret = epoll_ctl(efd_, EPOLL_CTL_DEL, (*fd_array_)[i].data.fd, nullptr);
                if (ret < 0)
                {
                    LOG(LogLevel::WARNING) << "移除事件失败";
                    exit(static_cast<int>(ErrorNumber::Epoll_Ctl_Fail));
                }
            }
            else
            {
                int ret = epoll_ctl(efd_, EPOLL_CTL_DEL, (*fd_array_)[i].data.fd, nullptr);
                if (ret < 0)
                {
                    LOG(LogLevel::WARNING) << "移除事件失败";
                    exit(static_cast<int>(ErrorNumber::Epoll_Ctl_Fail));
                }
            }
        }
    }
}

ssize_t recvData(std::string &out_data, int ac_socketfd) override
{
    char buffer[4096] = {0};
    ssize_t ret = recv(ac_socketfd, buffer, sizeof(buffer), 0);
    if (ret > 0)
        out_data = buffer;
    else
        close(ac_socketfd);

    return ret;
}

// 接收数据
ssize_t recvData(std::string &out_data, int ac_socketfd) override
{
    char buffer[4096] = {0};
    ssize_t ret = recv(ac_socketfd, buffer, sizeof(buffer), 0);
    if (ret > 0)
        out_data = buffer;

    return ret;
}

void handler(int n)
{
    for (int i = 0; i < n; i++)
    {
        // 保证一定是就绪事件
        // 判断当前是否是监听套接字
        // ...
        else 
        {
            // 不是监听套接字
            std::string buffer;
            ssize_t ret = bs_->recvData(buffer, (*fd_array_)[i].data.fd);
            // ...
            else if(ret == 0)
            {
                // ...
                // 在移除之后关闭文件描述符
                close((*fd_array_)[i].data.fd);
                if (ret < 0)
                {
                    LOG(LogLevel::WARNING) << "移除事件失败";
                    exit(static_cast<int>(ErrorNumber::Epoll_Ctl_Fail));
                }
            }
            else
            {
                // ...
                // 在移除之后关闭文件描述符
                close((*fd_array_)[i].data.fd);
                if (ret < 0)
                {
                    LOG(LogLevel::WARNING) << "移除事件失败";
                    exit(static_cast<int>(ErrorNumber::Epoll_Ctl_Fail));
                }
            }
        }
    }
}

class Connection
{
private:
    int fd_; // 当前套接字
    std::string in_buffer_; // 读取缓冲区
    std::string out_buffer_; // 写入缓冲区
    SockAddrIn client_; // 客户端信息
};

class Connection
{
public:
    // ...

    // 读、写和异常纯虚函数
    virtual void recvData() = 0;
    virtual void sendData() = 0;
    virtual void handleException() = 0;

    // ...
};

class Connection
{
// ...
// private:
protected:
// ...
};

class Connection
{
private:
    // ...

    uint32_t events_; // 事件类型
};

class Connection
{
private:
    // ...

    uint32_t revents_; // 就绪事件类型
};

Connection()
    :fd_(-1), events_(0), revents_(0)
{}

using conn_t = std::shared_ptr<Connection>;

class EpollServer
{
private:
    std::unordered_map<int, conn_t> connections_; // 文件描述符与Connection类对象指针的映射关系
};

class EpollServer
{
public:
    EpollServer()
        :revents_arr_(std::make_shared<std::array<struct epoll_event, g_default_array_num>>())
    {

    }

private:
    // ...
    std::shared_ptr<std::array<struct epoll_event, g_default_array_num>> revents_arr_; // 就绪事件数组
};

class Epoll
{
public:
    Epoll()
        :epfd_(-1)
    {
        epfd_ = epoll_create(256);
        if(epfd_ < 0)
        {
            LOG(LogLevel::ERROR) << "Epoll模型创建失败";
            exit(static_cast<int>(ErrorNumber::Epoll_Create_Fail));
        }

        LOG(LogLevel::INFO) << "Epoll模型创建成功";
    }

private:
    int epfd_;
};

class EpollServer
{
public:
    EpollServer()
        // ...
        ,ep_(std::make_shared<Epoll>())
    {

    }

private:
    // ...
    std::shared_ptr<Epoll> ep_; // 封装的Epoll类
};

void epollCtl(int fd, uint32_t events)
{
    struct epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    int ret = epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &ev);
    if(ret < 0)
    {
        LOG(LogLevel::WARNING) << "添加文件描述符和事件失败";
        exit(static_cast<int>(ErrorNumber::Epoll_Ctl_Fail));
    }

    LOG(LogLevel::INFO) << "添加文件描述符和事件成功";
}

// 添加文件描述符和关心的事件
void insertFdAndEvents(conn_t con)
{
    auto pos = connections_.find(con->getFd());
    if(pos != connections_.end())
    {
        LOG(LogLevel::WARNING) << "插入失败，指定描述符已存在";
        exit(static_cast<int>(ErrorNumber::Epoll_Ctl_Fail));
    }

    // 将指定的文件描述符和对应的连接对象建立映射关系
    connections_.insert({con->getFd(), con});

    // 添加到Epoll模型中
    ep_->epollCtl(con->getFd(), con->getEvents());
}

// 等待就绪事件
int epollWait(struct epoll_event* ep_arr, int maxSize, int timeout)
{
    int num = epoll_wait(epfd_, ep_arr, maxSize, timeout);

    if(num < 0)
    {
        LOG(LogLevel::WARNING) << "Epoll等待错误";
        exit(static_cast<int>(ErrorNumber::Epoll_Wait_Fail));
    }

    LOG(LogLevel::INFO) << "Epoll开始等待";

    return num;
}

class EpollServer
{
public:
    EpollServer()
        // ...
        ,isRunning_(false)
    {

    }

    // ...

    // 单次循环
    void loopOnce()
    {

    }

    // 启动服务器
    void startServer()
    {
        isRunning_ = true;
        while (isRunning_)
        {
            loopOnce();
        }

        isRunning_ = false;
    }

    ~EpollServer()
    {}

private:
    // ...
    bool isRunning_; // 服务器运行标识
};

// 停止服务器
void stopServer()
{
    isRunning_ = false;
}

// 判断文件描述符是否存在
bool checkFdIsInConnections(int fd)
{
    return connections_.find(fd) != connections_.end();
}

// 单次循环
void loopOnce()
{
    int timeout = 1000;
    int num = ep_->epollWait(revents_arr_->data(), g_default_array_num, timeout);
    for (int i = 0; i < num; i++)
    {
        // 获取当前文件描述符和就绪的事件
        int fd = (*revents_arr_)[i].data.fd;
        uint32_t revents = (*revents_arr_)[i].events;
        if ((revents & EPOLLERR) || (revents & EPOLLHUP))
            revents = (EPOLLIN | EPOLLOUT);
        if ((revents & EPOLLIN) && checkFdIsInConnections(fd))
        {
            // 执行读方法
            connections_[fd]->recvData();
        }
        if ((revents & EPOLLOUT) && checkFdIsInConnections(fd))
        {
            // 执行写方法
            connections_[fd]->sendData();
        }
    }
}

class Listener : public Connection
{
};

class Listener : public Connection
{
public:
    Listener(uint16_t port = default_port)
        :bs_(std::make_shared<TcpSocket>(port))
    {
    }

private:
    std::shared_ptr<BaseSocket> bs_;
};

Listener(uint16_t port = default_port)
    :bs_(std::make_shared<TcpSocket>(port))
{
    // 初始化套接字
    bs_->initSocket();
    // 调用父类函数设置监听套接字
    setFd(bs_->getListenSocketFd());
}

class EpollServerUtils
{
public:
    static void setNonBlock(int fd)
    {
        // 获取文件描述符已有的模式
        int mode = fcntl(fd, F_GETFL);
        if (mode < 0)
        {
            LOG(LogLevel::WARNING) << "获取文件描述符已有模式失败";
            exit(static_cast<int>(ErrorNumber::Fcntl_Fail));
        }

        // 设置非阻塞
        fcntl(fd, F_SETFL, mode | O_NONBLOCK);
    }
};

class BaseSocket
{
public:
    // ...
    // 获取监听套接字
    virtual int getListenSocketFd() = 0;

    // 具体实现方法
    void initSocket()
    {
        // 创建套接字
        createSocket();
        // 设置非阻塞
        EpollServerUtils::setNonBlock(getListenSocketFd());
        // ...
    }
};

// 接收信息
void recvData() override
{
    // 循环接收
    while(true)
    {
        int rerrno = 0;
        SockAddrIn client;
        int ac_socketfd = bs_->toAccept(std::addressof(client), &rerrno);
        // 当做IO行为处理
        if(ac_socketfd > 0)
        {
            // 正常情况，建立链接
        }
        else if(ac_socketfd < 0)
        {
            if(rerrno == EAGAIN || rerrno == EWOULDBLOCK)
            {
                LOG(LogLevel::INFO) << "accept接收结束";
                break;
            }
            else if(rerrno == EINTR)
            {
                LOG(LogLevel::INFO) << "accept被信号中断，重新接收";
                continue;
            }
            else
            {
                LOG(LogLevel::ERROR) << "accept错误";
                break;
            }
        }
    }
}

class IOService : public Connection
{
public:
    void recvData()
    {
    }

    void sendData()
    {
    }

    void handleException()
    {
    }
};

// Connection类所在文件
#include "EpollServer.h"

// Connection类所在文件
class EpollServer;

// Connection类所在文件
class EpollServer;

namespace connectionModule
{
    class Connection
    {
    // ...
    protected:
        // 指向EpollServer指针
        std::weak_ptr<EpollServer> ep_svr_;
        // 报错："connectionModule::EpollServer" is ambiguous
    };
}

// 前置声明
namespace epollServerModule
{
    class EpollServer;
}

namespace connectionModule
{
    class Connection
    {

    protected:
        // ...
        // 指向EpollServer指针
        std::weak_ptr<epollServerModule::EpollServer> ep_svr_;
        // ...
    };
}

Connection(std::shared_ptr<epollServerModule::EpollServer> ep)
    : fd_(-1), events_(0), revents_(0), ep_svr_(ep) 
{
}

Listener(std::shared_ptr<epollServerModule::EpollServer> ep,  uint16_t port = default_port)
    : Connection(ep), bs_(std::make_shared<TcpSocket>(port))
{
    // 初始化套接字
    bs_->initSocket();
    // 调用父类函数设置监听套接字
    setFd(bs_->getListenSocketFd());
}

IOService(std::shared_ptr<epollServerModule::EpollServer> ep)
    : Connection(ep)
{}

// 获取EpollServer指针
std::weak_ptr<epollServerModule::EpollServer> getEpollServer()
{
    return ep_svr_;
}

// 接收信息
void recvData() override
{
    // 循环接收
    while (true)
    {
        int rerrno = 0;
        SockAddrIn client;
        int ac_socketfd = bs_->toAccept(std::addressof(client), &rerrno);
        // 当做IO行为处理
        if (ac_socketfd > 0)
        {
            // 获取EpollServer类对象
            std::shared_ptr<epollServerModule::EpollServer> ep = getEpollServer().lock();
            // 正常情况，建立链接
            // 1. 创建连接对象
            std::shared_ptr<Connection> ac_con = std::make_shared<IOService>(ep);
            // 2. 设置非阻塞
            EpollServerUtils::setNonBlock(ac_socketfd);
            ac_con->setFd(ac_socketfd);
            ac_con->setEvent(EPOLLIN | EPOLLET);
            ac_con->setClient(std::move(client));
            // 3. 通过Connection将链接指针插入到Epoll模型中
            ep->insertFdAndEvents(ac_con);
        }
        // ...
    }
}

// 实现接收
int toAccept(SockAddrIn *client, int *out_errno) override
{
    // ...
    // 获取accept的错误码
    *out_errno = errno;

    // 删除下面的逻辑——起始
    if (ac_socketfd < 0)
    {
        LOG(LogLevel::ERROR) << "接收失败：" << strerror(errno);
        exit(static_cast<int>(ErrorNumber::AcceptFail));
    }
    // 结束

    // ...
}

void recvData()
{
    LOG(LogLevel::DEBUG) << "进入IOService读模块";
}

#include "Listener.hpp"
#include "EpollServer.hpp"

using namespace listenerModule;
using namespace epollServerModule;

int main()
{
    std::shared_ptr<epollServerModule::EpollServer> e_svr = std::make_shared<epollServerModule::EpollServer>();
    // 创建Listener代表启动服务器
    // 将Listener套接字构建为Connection对象
    std::shared_ptr<Connection> con = std::make_shared<Listener>(e_svr);
    con->setFd(con->getFd());
    // 使用EPOLLET开启边缘触发模式
    con->setEvent(EPOLLIN | EPOLLET);
    // 将Connection对象插入到Epoll模型中
    e_svr->insertFdAndEvents(con);
    // 启动服务器
    e_svr->startServer();

    return 0;
}