基于epoll的聊天室程序

http://blog.csdn.net/zhx6044/article/details/9320637

epoll相对于poll和select这两个多路复用的I/O模型更加的高效。epoll的函数很简单，麻烦的地方在于水平出发和边沿触发。

               用张图来说明下

              

ET(边沿)只是在状态反转时触发，比如从不可读到可读。而LT(水平)就是如果可读，就会一直触发。所以在使用ET的时候要做一些额外的处理，比如可读的，一直把缓冲区读完，进入不可读状态，下次来数据才会触发。

下面贴出代码，只是一个简单的练习的例子

#ifndef SOCKETHEADS_H
#define SOCKETHEADS_H


#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#endif //SOCKETHEADS_H


#ifndef EPOLL_H
#define EPOLL_H

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


/**
 * @brief The Epoll class 对epoll的封装
 */
class Epoll
{
public:
    /**
     *
     */
    enum EPOLL_OP {ADD = EPOLL_CTL_ADD, MOD = EPOLL_CTL_MOD, DEL = EPOLL_CTL_DEL};
    /**
     * 最大的连接数和最大的回传事件数
     */
    Epoll(int _max = 30, int maxevents = 20);
    ~Epoll();
    int create();
    int add(int fd, epoll_event *event);
    int mod(int fd, epoll_event *event);
    int del(int fd, epoll_event *event);
    void setTimeout(int timeout);
    void setMaxEvents(int maxevents);
    int wait();
    const epoll_event* events() const;
    const epoll_event& operator[](int index)
    {
        return backEvents[index];
    }
private:
    bool isValid() const;
    int max;
    int epoll_fd;
    int epoll_timeout;
    int epoll_maxevents;
    epoll_event *backEvents;
};




#endif //EPOLL_H

#include "zepoll.h"


Epoll::Epoll(int _max, int maxevents):max(_max),
    epoll_fd(-1),
    epoll_timeout(0),
    epoll_maxevents(maxevents),
    backEvents(0)
{

}

Epoll::~Epoll()
{
    if (isValid()) {
        close(epoll_fd);
    }
    delete[] backEvents;
}


inline
bool Epoll::isValid() const
{
    return epoll_fd > 0;
}

inline
void Epoll::setTimeout(int timeout)
{
    epoll_timeout = timeout;
}

inline
void Epoll::setMaxEvents(int maxevents)
{
    epoll_maxevents = maxevents;
}

inline
const epoll_event* Epoll::events() const
{
    return backEvents;
}



int Epoll::create()
{
    epoll_fd = ::epoll_create(max);
    if (isValid()) {
        backEvents = new epoll_event[epoll_maxevents];
    }
    return epoll_fd;
}

int Epoll::add(int fd, epoll_event *event)
{
    if (isValid()) {
        return ::epoll_ctl(epoll_fd, ADD, fd, event);
    }
    return -1;

}

int Epoll::mod(int fd, epoll_event *event)
{
    if (isValid()) {
        return ::epoll_ctl(epoll_fd, MOD, fd, event);
    }
    return -1;

}

int Epoll::del(int fd, epoll_event *event)
{
    if (isValid()) {
        return ::epoll_ctl(epoll_fd, DEL, fd, event);
    }
    return -1;
}

int Epoll::wait()
{
    if (isValid()) {
        return ::epoll_wait(epoll_fd, backEvents, epoll_maxevents, epoll_timeout);
    }
    return -1;
}

/********************************************************************
* author 周翔
* e-mail [e-mail]
* blog http://blog.csdn.net/zhx6044
**********************************************************************/

#ifndef TASK_H
#define TASK_H

#include <string>
#include <socketheads.h>

/**
 * @brief The Task class 任务类
 */
class Task
{
public:
    typedef enum {CONNECT = 0, DISCONNECT, TALKING} TASKFLAG;
    Task(const std::string &message, TASKFLAG flag = TALKING);
    const std::string& getMessage() const;
    TASKFLAG getFlag() const;
    void setIP(in_addr _ip);
    int getS_fd() const;

    void setS_fd(int _fd);


    std::string getData() const;


private:
    std::string m_message;
    TASKFLAG m_flag;
    in_addr ip;
    int s_fd;
};

#endif // TASK_H

/********************************************************************
* author 周翔
* e-mail [e-mail]
* blog http://blog.csdn.net/zhx6044
**********************************************************************/

#include "task.h"

Task::Task(const std::string &message, TASKFLAG flag):
    m_message(message),
    m_flag(flag)
{
}


const std::string& Task::getMessage() const
{
    return m_message;
}

Task::TASKFLAG Task::getFlag() const
{
    return m_flag;
}

void Task::setIP(in_addr _ip)
{
    ip = _ip;
}

int Task::getS_fd() const
{
    return s_fd;
}

void Task::setS_fd(int _fd)
{
    s_fd = _fd;
}

std::string Task::getData() const
{
    std::string re;
    if (m_flag == CONNECT) {
        re = ::inet_ntoa(ip) + std::string("----->") + "CONNECT!    " + m_message;
    } else {
        if (m_flag == DISCONNECT) {
            re = ::inet_ntoa(ip) + std::string("----->") + "DISCONNECT   " + m_message;;
        } else {
            re = ::inet_ntoa(ip) + std::string("----->Talk:") + m_message;
        }
    }
    return re;
}

#ifndef EPOLL_SERVER_H
#define EPOLL_SERVER_H

#include <map>
#include <list>

#include "zepoll.h"
#include "socketheads.h"
#include "task.h"


typedef std::pair<int, in_addr> FDtoIP;

/**
 * @brief The Epoll_server class 服务器
 */
class Epoll_server
{
public:
    Epoll_server(int port);
    ~Epoll_server();
    int bind();
    int listen();
    void poweroff();
    bool states() const;
private:
    enum {BLOCKLOG = 5};

    bool isValid() const;

    int acceptSocketEpoll();
    int readSocketEpoll(const epoll_event &ev);
    int writeSocketEpoll(const epoll_event &ev);

    void doTask(const Task &t);

    int _port;
    int server_socket_fd;
    Epoll *_epoll;
    sockaddr_in server_addr;
    sockaddr_in client_addr;
    epoll_event m_event;
    bool on;


    static int setNonblocking(int socket_fd);


    std::list<FDtoIP> fd_IP;




};

#endif //EPOLL_SERVER_H

#include "epoll_server.h"

#include <iostream>


//static char welcom[] = "welcom to my epoll_server";
//static char sorry[] = "Sorry! This is a simple demo,so not any function!";
//static char buf[BUFSIZ];


Epoll_server::Epoll_server(int port):_port(port),
    server_socket_fd(-1),
    _epoll(0),
    on(true)
{


}


Epoll_server::~Epoll_server()
{
    if (isValid()) {
        ::close(server_socket_fd);
    }
    delete _epoll;

}


inline
bool Epoll_server::isValid() const
{
    return server_socket_fd > 0;
}


inline
void Epoll_server::poweroff()
{
    on = false;
}



inline
bool Epoll_server::states() const
{
    return on;
}



int Epoll_server::setNonblocking(int socket_fd)
{
    int opts;
    opts = fcntl(socket_fd, F_GETFL);
    if (opts < 0) {
        return -1;
    } else
    {
        opts = opts | O_NONBLOCK;
        if (fcntl(socket_fd, F_SETFL, opts) < 0) {

            return -1;
        }
    }
    return 0;
}


void Epoll_server::doTask(const Task &t)
{
    std::list<FDtoIP>::iterator ite = fd_IP.begin();
    std::list<FDtoIP>::iterator ite1 = fd_IP.end();
    for (;ite != fd_IP.end();++ite) {
        if ((*ite).first != t.getS_fd()) {
            memset(&m_event, '\0', sizeof(m_event));
            m_event.events = EPOLLOUT | EPOLLET;
            Task *c = new Task(t);
            c->setS_fd((*ite).first);
            m_event.data.ptr = static_cast<void*>(c);
            _epoll->mod((*ite).first, &m_event);
        } else {
            ite1 = ite;
        }
    }
    if (t.getFlag() == Task::DISCONNECT) {
        if (ite1 != fd_IP.end()) {
            fd_IP.erase(ite1);
        }

    }

}
/**
 * @brief Epoll_server::acceptSocketEpoll 有用户接入
 * @return
 */
int Epoll_server::acceptSocketEpoll()
{
    socklen_t len = sizeof(struct sockaddr_in);
    int connect_fd;
    while ((connect_fd = ::accept(server_socket_fd,
                                  (struct sockaddr*)(&client_addr), &len)) > 0) {
        if (setNonblocking(connect_fd) < 0) {
            ::close(connect_fd);
            continue;
        }
        m_event.data.fd = connect_fd;
        m_event.events = EPOLLIN | EPOLLET;
        if (_epoll->add(connect_fd, &m_event) < 0) {
            ::close(connect_fd);
            continue;
        } else {

            fd_IP.push_back(FDtoIP(connect_fd, client_addr.sin_addr));
            Task t("come in", Task::CONNECT);
            t.setIP(client_addr.sin_addr);
            t.setS_fd(connect_fd);
            doTask(t);

        }



    }

    if (connect_fd == -1 && errno != EAGAIN && errno != ECONNABORTED
            && errno != EPROTO && errno !=EINTR) {
        return -1;

    }



    return 0;


}


int Epoll_server::readSocketEpoll(const epoll_event &ev)
{
    int n = 0;
    int nread = 0;
    char buf[BUFSIZ]={'\0'};
    while ((nread = ::read(ev.data.fd, buf + n, BUFSIZ-1)) > 0) {
        n += nread;
    }
    if (nread == -1 && errno != EAGAIN) {
        return -1;
    }

    std::list<FDtoIP>::iterator ite = fd_IP.begin();
    for (;ite != fd_IP.end();++ite) {
        if ((*ite).first == ev.data.fd) {
            break;
        }
    }

    if (nread == 0) {
        strcpy(buf, " disconet  left ");
        Task t(buf,Task::DISCONNECT);
        t.setIP(client_addr.sin_addr);
        t.setS_fd((*ite).first);
        doTask(t);
    } else {
        Task t(buf,Task::TALKING);
        t.setIP(client_addr.sin_addr);
        t.setS_fd((*ite).first);
        doTask(t);
    }

    // Task *t = new Task(buf,Task::DISCONNECT);
    // t->setIP((*ite).second);
    // t->setS_fd((*ite).first);




    // m_event.data.fd = ev.data.fd;
    // m_event.events = ev.events;
    return 0;//_epoll->mod(m_event.data.fd, &m_event);
}


int Epoll_server::writeSocketEpoll(const epoll_event &ev)
{
    Task *t = static_cast<Task*>(ev.data.ptr);
    const char* buf = t->getData().data();
    int nwrite = 0, data_size = strlen(buf);
    int fd = t->getS_fd();
    int n = data_size;
    delete t;
    while (n > 0) {
        nwrite = ::write(fd, buf + data_size - n, n);
        if (nwrite < 0) {
            if (nwrite == -1 && errno != EAGAIN) {
                return -1;
            }
            break;
        }
        n -= nwrite;
    }

    memset(&m_event, '\0', sizeof(m_event));
    // m_event.events &= ~EPOLLOUT;
    m_event.events = EPOLLIN | EPOLLET;
    m_event.data.fd = fd;
    if (_epoll->mod(fd, &m_event) < 0) {
        ::close(m_event.data.fd);
        return -1;
    }
    return 0;
}

/**
 * @brief Epoll_server::bind
 * @return
 */
int Epoll_server::bind()
{
    server_socket_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (server_socket_fd < 0) {
        return -1;
    }

    setNonblocking(server_socket_fd);

    _epoll = new Epoll();
    if (_epoll->create() < 0) {
        return -1;
    }
    memset(&m_event, '\0', sizeof(m_event));
    m_event.data.fd = server_socket_fd;
    m_event.events = EPOLLIN | EPOLLET;
    _epoll->add(server_socket_fd, &m_event);

    memset(&server_addr, 0 ,sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(_port);

    return ::bind(server_socket_fd, (struct sockaddr*)(&server_addr), sizeof(struct sockaddr));

}


int Epoll_server::listen()
{
    if (isValid()) {
        if (::listen(server_socket_fd, BLOCKLOG) < 0) {
            return -1;
        } else {
            int num;
            while (on) {
                num = _epoll->wait();
                for (int i = 0;i < num;++i) {
                    /**
                     * 接受连接的连接，把她加入到epoll中
                     */
                    if ((*_epoll)[i].data.fd == server_socket_fd) {
                        if (acceptSocketEpoll() < 0) {
                            break;
                        }
                        continue;

                    }
                    /**
                     * EPOLLIN event
                     */
                    if ((*_epoll)[i].events & EPOLLIN) {
                        if (readSocketEpoll((*_epoll)[i]) < 0) {
                            break;
                        }
                        continue;


                    }

                    /**
                     * EPOLLOUT event
                     */
                    if ((*_epoll)[i].events & EPOLLOUT) {
                        if (writeSocketEpoll((*_epoll)[i]) < 0) {
                            break;
                        }

                    }
                }
            }
        }
    }
    return -1;
}


#include "epoll_server.h"

#include <iostream>



int main(int /*argc*/, char const **/*argv[]*/)
{
    std::cout << "server" << std::endl;
    Epoll_server s(18090);
    if (s.bind() < 0) {
        return -1;
    }
    return s.listen();
}

需要客户端程序的在这里