epoll极简介绍

• 关于epoll的详细介绍，已经有较多文章可以参考，例如这篇文章介绍就比较详细：
  http://blog.chinaunix.net/uid-24517549-id-4051156.html

• epoll编程的接口：

  • epoll_create 创建一个epoll内核对象，返回指向该对象的fd
  • epoll_ctl 往epoll中添加、删除、修改需要监控的套接字
  • epoll_wait 等待epoll中的套接字产生可读、可写、异常消息

• 使用epoll时有如下应该注意的地方：

  • 确保被epoll的套接字必须是非阻塞的
  • 读取可读的TCP套接字时，需要在循环中读取多次，直到返回值为-1且errno为EAGAIN为止，因为只有这种情况才说明可读的数据已经全部读完了
  • 已经出错或断开连接的fd需要及时从epoll中删除掉，然后close fd

如下代码是使用epoll实现的多路复用TCP的简单Server及其测试客户端：

/******************************************************************************
 * 文件名称：TestEpoll.cpp
 * 文件描述：Epoll测试服务器端
 * 创建日期：2015-04-09
 * 作    者：casheywen
 ******************************************************************************/

#include <iostream>
using namespace std;
#include <errno.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>

#define LOG_ERR(fmt, args...) fprintf(stderr, "%d|"fmt"\n", __LINE__, ##args)
#define LOG_INFO(fmt, args...) fprintf(stdout, "%d|"fmt"\n", __LINE__, ##args)

int CreateListenFd(const char *pszIP, uint16_t usPort)
{
    struct sockaddr_in stAddr;
    stAddr.sin_family = AF_INET;
    stAddr.sin_port = htons(usPort);
    stAddr.sin_addr.s_addr = inet_addr(pszIP);
    socklen_t nAddrLen = sizeof(struct sockaddr_in);

    int iFd = socket(AF_INET, SOCK_STREAM, 0);

    if (iFd < 0)
    {
        LOG_ERR("create socket fail: %s", strerror(errno)); 
        return -1;
    }

    if (0 > bind(iFd, (struct sockaddr *)&stAddr, nAddrLen))
    {
        LOG_ERR("bind fail: %s", strerror(errno));
        return -1;
    }

    if (0 > listen(iFd, 64))
    {
        LOG_ERR("listen fail: %s", strerror(errno));
        return -1;
    }

    LOG_INFO("Listening: %s:%hu, fd=%d", pszIP, usPort, iFd);

    return iFd;
}

bool SetSockNonBlock(int iSockfd)
{
    int iRet = fcntl(iSockfd, F_GETFL, 0);
    if (-1 == iRet)
    {
        return false;    
    }

    if (-1 == fcntl(iSockfd, F_SETFL, iRet | O_NONBLOCK))
    {
        return false;
    }

    return true;
}

int main()
{
    int iEpollFd = epoll_create(100);    // 100为预估需要epoll的fd数量
    if (iEpollFd < 0)
    {
        LOG_ERR("epoll_create fail: %s");
        return 1;
    }

    int iListenFd = CreateListenFd("0.0.0.0", 12333);
    if (iListenFd < 0)
    {
        LOG_ERR("CreateListenFd Fail");
        return 1;
    }

    if (!SetSockNonBlock(iListenFd))    // 确保socket为非阻塞状态
    {
        LOG_ERR("SetSockNonBlock Fail: %s", strerror(errno));
        return 1;
    }

    struct epoll_event ev, events[20];

    ev.events = EPOLLIN;
    ev.data.fd = iListenFd;

    // 将监听的socket加入epoll
    int iRet = epoll_ctl(iEpollFd, EPOLL_CTL_ADD, iListenFd, &ev);
    if (iRet < 0)
    {
        LOG_ERR("epoll_ctl fail: %s", strerror(errno)); 
        return 1;
    }

    while (true) 
    {
        iRet = epoll_wait(iEpollFd, events, sizeof(events), -1);    // 最后的-1表示超时时间无穷大

        if (iRet < 0)
        {
            if (errno == EINTR)
            {
                LOG_ERR("Interrupted, quit.");
            }
            else
            {
                LOG_ERR("epoll_wait fail: %s", strerror(errno)); 
            }
            return 1;
        }

        int iEvents = iRet;

        for (int i = 0; i < iEvents; i++)
        {
            // 对于监听状态中的套接字，可读意味着有新的连接
            if (events[i].data.fd == iListenFd)
            {
                struct sockaddr_in stClientAddr;
                socklen_t nAddrLen = sizeof(stClientAddr);

                memset(&stClientAddr, 0, sizeof(stClientAddr));

                int iClientFd = accept(iListenFd, (struct sockaddr *)&stClientAddr, &nAddrLen);
                if (iClientFd < 0)
                {
                    LOG_ERR("accept fail: %s", strerror(errno));
                    return 1;
                }

                if (!SetSockNonBlock(iClientFd))    // 将新连接的套接字设置为非阻塞
                {
                    LOG_ERR("SetSockNonBlock Fail: fd=%d %s", iClientFd, strerror(errno));
                    return 1;
                }

                LOG_INFO("Connected:%s:%hu, fd=%d", inet_ntoa(stClientAddr.sin_addr), htons(stClientAddr.sin_port), iClientFd);

                ev.events = EPOLLIN;
                ev.data.fd = iClientFd;

                // 将连接的fd加入epoll
                int iRet = epoll_ctl(iEpollFd, EPOLL_CTL_ADD, iClientFd, &ev);
                if (iRet < 0)
                {
                    LOG_ERR("epoll_ctl fail: %s", strerror(errno)); 
                    return 1;
                }
            }
            else   // 对于客户端连接可读的情况
            {
                int iClientFd = events[i].data.fd;

                static char s_acBuf[10 * 1024] = {0};

                int iTotal = 0;

                do
                {
                    iRet = recv(iClientFd, &s_acBuf[iTotal], sizeof(s_acBuf) - iTotal, 0);

                    if (iRet > 0)
                    {
                        iTotal += iRet;
                    }
                    else if (iRet < 0 && errno == EAGAIN)
                    {
                        LOG_INFO("Total: %d Bytes, [%s]", iTotal, s_acBuf);
                        break;
                    }
                    else
                    {
                        if (iRet == 0)    // 连接已经断开
                        {
                            LOG_INFO("Disconnected: fd=%d", iClientFd);
                        }
                        else // iRet < 0 出现错误
                        {
                            LOG_INFO("recv fail: fd=%d %s", iClientFd, strerror(errno));
                        }

                        // 将出错或断开连接的fd从epoll中去掉
                        int iRet = epoll_ctl(iEpollFd, EPOLL_CTL_DEL, iClientFd, NULL);
                        if (iRet < 0)
                        {
                            LOG_ERR("epoll_ctl fail: %s", strerror(errno)); 
                            return 1;
                        }

                        close(iClientFd);
                        break;
                    }

                } while (iTotal < sizeof(s_acBuf));
            }
        }
    }

    return 0;
}

/******************************************************************************
 * 文件名称：TcpClient.cpp
 * 文件描述：Epoll测试客户端
 * 创建日期：2015-04-09
 * 作    者：casheywen
 ******************************************************************************/

#include <iostream>
using namespace std;
#include <errno.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <signal.h>

#define LOG_ERR(fmt, args...) fprintf(stderr, "%d|"fmt"\n", __LINE__, ##args)
#define LOG_INFO(fmt, args...) fprintf(stdout, "%d|"fmt"\n", __LINE__, ##args)

void SigPipeHandler(int iSigno)
{
    LOG_ERR("SigPipe received");
    exit(1);
}

bool ConnectTcpSocket(int iFd, const char *pszIP, uint16_t usPort)
{
    struct sockaddr_in stAddr;
    memset(&stAddr, 0, sizeof(stAddr));

    stAddr.sin_family = AF_INET;
    inet_aton(pszIP, &stAddr.sin_addr);
    stAddr.sin_port = htons(usPort);

    int iRet = connect(iFd, (struct sockaddr *)&stAddr, sizeof(stAddr));
    if (iRet < 0)
    {
        LOG_ERR("Connect Fail: %s", strerror(errno));
        return false;
    }

    return true;
}

int main()
{

    int iFd = socket(AF_INET, SOCK_STREAM, 0);
    if (iFd < 0)
    {
        LOG_ERR("Create Socket Fail: %s", strerror(errno));
        return 1;
    }

    if (!ConnectTcpSocket(iFd, "127.0.0.1", 12333))
    {
        LOG_ERR("ConnectUnixSocket Fail");
        return 1;
    }

    LOG_INFO("Connect Success");

    if (SIG_ERR == signal(SIGPIPE, SigPipeHandler))    // 当连接中断时调用write函数会收到SIGPIPE信号
    {
        LOG_ERR("Signal Fail: %s", strerror(errno));
        return 1;
    }

    char szContent[4096];
    ssize_t nWrite = 0;

    while (cin >> szContent)
    {
        nWrite = write(iFd, szContent, strlen(szContent));

        if (nWrite < 0)
        {
            LOG_ERR("write Fail: %s", strerror(errno));
            return 1;
        }
    }

    return 0;
}