Chinaunix首页 | 论坛 | 博客

justjust.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

justscu

  • 博客访问： 157511
  • 博文数量： 22
  • 博客积分： 425
  • 博客等级： 下士
  • 技术积分： 350
  • 用 户 组： 普通用户
  • 注册时间： 2012-05-15 09:43
个人简介

专注服务器设计、开发； https://github.com/justscu；

文章分类

全部博文（22）

文章存档

2014年（10）

2013年（2）

2012年（10）

我的朋友
最近访客
推荐博文
相关博文
用epoll实现一个异步处理的模型(1)

分类： LINUX

2013-11-06 20:50:45

1. 同select比较
    对于select，linux/posix_types.h头文件有这样的宏：#define __FD_SETSIZE  1024；
    select能监听的最大fd数量，在内核编译的时候就确定了的。若想加大fd的数量，需要调整编译参数，重新编译内核。
    epoll_wait直接返回被触发的fd或对应的一块buffer，不需要遍历所有的fd。
    epoll的实现中，用户空间跟内核空间共享内存(mmap)，避免拷贝。

2. 基本的数据结构
    typedef union epoll_data { //注意是union
       void *ptr;
       int fd;
       uint32_t u32;
       uint64_t u64;
   } epoll_data_t;

    struct epoll_event {
       uint32_t events; /* Epoll events */
       epoll_data_t data; /* User data variable */
    };

3. 基本操作
3.1 加入到epoll监控
    struct epoll_event ee;
    ee.events = EPOLLIN|EPOLLOUT|EPOLLET; //不设置EPOLLET的话，为LT模式
    ee.data.ptr = (void *) buffer;//传入buffer地址
    epoll_ctl(epfd, EPOLL_CTL_ADD, socketfd, &ee);

3.2 从epoll监控中删除
   struct epoll_event ee;
   ee.events = 0;
   ee.data.ptr = NULL;
   epoll_ctl(epfd, EPOLL_CTL_DEL, socketfd, &ee);

3.3 等待事件的触发
    for (;;){
        nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
        if (nfds == -1){    
            if (errno == EINTR)
                continue; // 中断
            perror("epoll_pwait");
            exit (EXIT_FAILURE);
        }

        for (n = 0; n < nfds; ++n){
            // events[n].events; //是什么事件 EPOLLIN/EPOLLOUT/EPOLLERR/EPOLLPRI
            // events[n].data.fd;//被触发的fd(在epoll_ctrl时传入的)
            // events[n].data.ptr;//跟fd相关联的buffer(在epoll_ctrl时传入的)
            if( events[n].events & EPOLLIN ) { }
            else if( events[n].events & EPOLLOUT ) { }
            else if( events[n].events & EPOLLERR ) { }
            else {}
        }
}

4. 读和写
    epoll有2种模式， 水平触发（LT）&边缘触发（ET）。
4.1 LT模式下的读和写
    读：若接收缓冲区中有数据，就会一直触发EPOLLIN。所以不用担心是否读干净缓冲区中的数据，若没有读完的话，EPOLL会一直被触发。
    写：若发送缓冲区中还有空间的话，会一直触发EPOLLOUT（只有在发送缓冲区满的时候，才不会被触发）。所以这个很烦，本来没有数据需要发送，但EPOLLOUT仍然被触发。解决方法：
    （1）epoll_ctl(, EPOLL_CTL_ADD, fd)，将该fd加入到epoll中；
    （2）该fd的EPOLLOUT被触发，发送数据（一般在回调函数中完成）；
    （3）epoll_ctl(epfd, EPOLL_CTL_DEL, fd),将该fd移出epoll。加入又移出的，很麻烦，又有开销。若数据比较少的时候，直接调用send发送好了。
4.2 ET模式下的读和写
    ET模式下，是不能用阻塞式的函数的。
    读：fd由不可读->可读（接收缓冲区没有数据->有数据）时，才会触发EPOLLIN。若没有读干净，read返回，缓冲区仍然有数据时， 不会再次触发。
    n=read(fd, buf, buf_size);
    if(n<0){
        if (errno == EAGAIN)
            continue;
        else
            return -1; //errno, should close the fd
    }
    else if(n < buf_size){
        // 读完了缓冲区中的数据，可以返回
        return n;
    }
    else if(n == buf_size){
        // 缓冲区中可能还有数据，继续读
        continue;
    }
    写：fd由不可写->可写（发送缓冲区满->有空间）时，才会触发EPOLLOUT。

 5. 一个异步模型
    （1）将要监控的fd加入到epoll的监控中，每一个fd对应一个事件处理类(EventHanlder)，当fd被触发时，处理类中的相关函数被调用。
    （2）用epoll可以实现定时器。依赖于epoll_wait的超时机制。该定时器存在误差。

6. 代码

reactor.h

  1. #pragma once

  3. #include <time.h>
  4. #include <map>
  5. #include <sys/time.h>
  6. #include <fcntl.h>
  7. #include "min_heap.h"

  9. class FDEventHanderBase;
  10. class FDEventHanderSet;
  11. class TimerEventHanderBase;

  13. class NonCopyable
  14. {
  15. protected:
  16.     NonCopyable() {
  17.     }
  18.     virtual ~NonCopyable() {
  19.     }
  20. private:
  21.     NonCopyable(const NonCopyable &); //禁止拷贝
  22.     const NonCopyable & operator=(const NonCopyable &); //禁止赋值
  23. };

  25. class Reactor: public NonCopyable
  26. {
  27. public:
  28.     Reactor() {
  29.     }

  31.     virtual ~Reactor() {
  32.     }

  34.     virtual bool Init() = 0;
  35.     virtual bool Run() = 0;
  36.     virtual bool Stop() = 0;
  37.     virtual int RegisterReadEvent(FDEventHanderBase* pBase) = 0;
  38.     virtual int RegisterWriteEvent(FDEventHanderBase* pBase) = 0;
  39.     virtual int UnRegisterReadEvent(FDEventHanderBase* pBase) = 0;
  40.     virtual int UnRegisterWriteEvent(FDEventHanderBase* pBase) = 0;
  41.     virtual int RegisterTimer(TimerEventHanderBase* pBase) = 0;
  42.     virtual int UnRegisterTimer(TimerEventHanderBase* pBase) = 0;
  43. };
  44. /* **********************************
  45.  *             处理FD类的基类
  46.  * 当该fd被触发时，调用处理函数来进行处理
  47.  * ***********************************/
  48. class FDEventHanderBase: public NonCopyable
  49. {
  50. public:
  51.     FDEventHanderBase(int fd, Reactor* pReactor) :
  52.         m_fd(fd), m_pReactor(pReactor)
  53.     {
  54.     }
  55.     virtual ~FDEventHanderBase() {
  56.     }
  57.     // 读被触发时，调用该函数
  58.     virtual void OnFDRead() = 0;
  59.     // 写被触发时，调用该函数
  60.     virtual void OnFDWrite() = 0;
  61.     // 关闭fd
  62.     virtual void Close() = 0;
  63.     // 注册读事件
  64.     int RegisterReadEvent();
  65.     // 注册写事件
  66.     int RegisterWriteEvent();
  67.     // 注销读事件
  68.     int UnRegisterReadEvent();
  69.     // 注销写事件
  70.     int UnRegisterWriteEvent();
  71.     int GetFD() const;
  72.     void SetFD(int fd);
  73.     // 设置非阻塞，返回0，success.
  74.     int SetNonBlock(int fd, bool bNonBlock = true);
  75. protected:
  76.     int m_fd;//该handler所对应的fd.
  77.     Reactor* m_pReactor;
  78. };
  79. /* **********************************
  80.  *        处理timer类的基类
  81.  *    当timer被触发时，调用该类的函数
  82.  * **********************************/
  83. class TimerEventHanderBase: public NonCopyable
  84. {
  85. public:
  86.     TimerEventHanderBase(Reactor* reactor) :
  87.         m_pReactor(reactor), m_bRestart(false)
  88.     {
  89.         timerclear(&m_interval);
  90.         timerclear(&m_endtime);
  91.     }
  92.     virtual ~TimerEventHanderBase() {
  93.     }
  94.     virtual void OnTimeOut() = 0;
  95.     // 注册定时器，@msec，多长时间响一次,单位:毫秒
  96.     // 由于epoll精度的限制，这个地方，精度只能达到 毫秒，但不是准确
  97.     int RegisterTimer(unsigned int msec /*ms*/, bool restart = false);
  98.     int UnRegisterTimer();
  99.     int RegisterTimerAgain();
  100.     // 获取结束时间
  101.     const timeval& GetEndTime() const {
  102.         return m_endtime;
  103.     }
  104.     // 该定时器是否需要重启
  105.     bool IsRestart() const {
  106.         return m_bRestart;
  107.     }
  108. protected:
  109.     Reactor* m_pReactor;
  110.     timeval m_interval;// 时间间隔
  111.     timeval m_endtime; // 定时器到期时间
  112.     bool m_bRestart;// 是否自动重启
  113. };

  115. /* **********************************
  116.  *             处理类的集合
  117.  *     有个对应关系 -- fd -- 处理函数 -- read/write.
  118.  * *********************************/
  119. class EventHanderSet
  120. {
  121. public:
  122.     // 增加fd事件, @type : EPOLLIN, EPOLLOUT，
  123.     void AddFDEventHandler(FDEventHanderBase* pHandler, int type);
  124.     // @type : EPOLLIN, EPOLLOUT
  125.     void DelFDEventHandler(int fd, int type);
  126.     // 根据fd，找到 FDEventHanderBase，及type.
  127.     FDEventHanderBase* GetFDEventHandler(int fd, int& type);
  128. public:
  129.     void AddTimerEventHandler(TimerEventHanderBase* pHandler);
  130.     void DelTimerEventHandler(TimerEventHanderBase* pHandler);
  131.     void ScanTimer();
  132. private:
  133.     struct FDHandler
  134.     {
  135.         int type; //EPOLLIN, EPOLLOUT
  136.         FDEventHanderBase* pHandler;
  137.     };
  138.     std::map<int, FDHandler> m_fdmap; // fd -- type -- handler

  140.     // 用于保存timer，在同一时刻，可能有多个handler.
  141.     MinHeap<std::pair<long int, long int> > m_timerMinHeap;
  142.     std::multimap<std::pair<long int, long int>, TimerEventHanderBase*>
  143.      m_timerMultiMap;
  144. };

  146. //////////// LT epoll /////////////////
  147. class LTReactor: public Reactor
  148. {
  149. public:
  150.     LTReactor();
  151.     virtual ~LTReactor();

  153.     bool Init();
  154.     bool Run();
  155.     bool Stop();
  156.     //注册读事件
  157.     int RegisterReadEvent(FDEventHanderBase* pBase);
  158.     //注册写事件
  159.     int RegisterWriteEvent(FDEventHanderBase* pBase);

  161.     int UnRegisterReadEvent(FDEventHanderBase* pBase);
  162.     int UnRegisterWriteEvent(FDEventHanderBase* pBase);
  163.     int UnRegisterAllEvent(FDEventHanderBase* pBase);
  164.     // 注册定时器
  165.     int RegisterTimer(TimerEventHanderBase* pBase);
  166.     int UnRegisterTimer(TimerEventHanderBase* pBase);
  167. private:
  168.     void ScanTimer();
  169. private:
  170.     int m_epfd; //epoll's fd
  171.     bool m_bRunning;
  172.     timeval m_stoptime; //停止时间
  173.     EventHanderSet m_handlerSet;
  174. };

reactor.cpp

  1. #include "reactor.h"
  2. #include <sys/epoll.h>
  3. #include <errno.h>
  4. #include <iostream>
  5. #include <string.h>
  6. #include <stdio.h>

  8. /////////////////////////////////////////////////////////
  9. #define MAX_EPOLL_SIZE 1024
  10. #define MAX_EPOLL_EVENTS_SIZE 128

  12. LTReactor::LTReactor() :
  13.     m_epfd(-1), m_bRunning(false)
  14. {
  15. }

  17. LTReactor::~LTReactor() {
  18. }

  20. bool LTReactor::Init() {
  21.     m_epfd = epoll_create(MAX_EPOLL_SIZE);
  22.     if (m_epfd == -1)
  23.     {
  24.         std::cout << "epoll_create error: " << strerror(errno) << std::endl;
  25.         return false;
  26.     }

  28.     m_bRunning = true;
  29.     return true;
  30. }

  32. bool LTReactor::Run() {
  33.     struct epoll_event events[MAX_EPOLL_EVENTS_SIZE];
  34.     int nfds = 0; //返回被触发的事件的个数
  35.     while (true)
  36.     {
  37.         // 超时时间，若m_bRunning，立即返回；否则100ms，即0.1s. 实际上，epoll_wait，只能够精确到毫秒(1/1000)
  38.         nfds = epoll_wait(m_epfd, events, sizeof(events) / sizeof(events[0]),
  39.          m_bRunning ? 100 : 1);
  40.         // 要停止epoll.
  41.         if (m_bRunning == false)
  42.         {
  43.             timeval now;
  44.             gettimeofday(&now, NULL); //获取当前时间
  45.             if (timercmp(&now, &m_stoptime, >))
  46.             {
  47.                 break;
  48.             }
  49.         }
  50.         // epoll 出错
  51.         if (nfds == -1)
  52.         {
  53.             std::cout << "epoll_wait error: " << strerror(errno) << std::endl;
  54.             continue;
  55.         }
  56.         // 处理被触发的事件
  57.         for (int i = 0; i < nfds; ++i)
  58.         {
  59.             int type = 0;
  60.             int fd = events[i].data.fd;
  61.             if (fd < 0) //fd出错
  62.             {
  63.                 std::cout << "the FD is " << fd << std::endl;
  64.                 continue;
  65.             }
  66.             FDEventHanderBase* pBase = m_handlerSet.GetFDEventHandler(fd, type);
  67.             if (NULL == pBase)
  68.             {
  69.                 std::cout << "pBase is NULL, fd is" << fd << std::endl;
  70.                 continue;
  71.             }

  73.             if (events[i].events & EPOLLIN)//read events
  74.             {
  75.                 pBase->OnFDRead();
  76.             }
  77.             if (events[i].events & EPOLLOUT) //write events
  78.             {
  79.                 pBase->OnFDWrite();
  80.             }
  81.             if (events[i].events & EPOLLERR) // error events
  82.             {
  83.                 pBase->Close();
  84.             }
  85.         }
  86.         // 处理定时器事件
  87.         ScanTimer();
  88.         // 处理idle事件
  89.     } // end of while

  91.     std::cout << "stop epoll" << std::endl;
  92.     close(m_epfd); //close epoll fd
  93.     m_epfd = -1;

  95.     return true;
  96. }

  98. bool LTReactor::Stop() {
  99.     m_bRunning = false;

  101.     timeval val, now;
  102.     val.tv_usec = 50 * 1000; //50ms后停止
  103.     gettimeofday(&now, NULL);
  104.     timeradd(&val, &now, &m_stoptime);
  105.     return true;
  106. }
  107. // 注册读事件,
  108. // @return: 0, success. else failed.
  109. int LTReactor::RegisterReadEvent(FDEventHanderBase* pBase) {
  110.     int type = 0;
  111.     FDEventHanderBase* pHander = m_handlerSet.GetFDEventHandler(pBase->GetFD(),
  112.      type);

  114.     epoll_event event;
  115.     event.data.fd = pBase->GetFD();
  116.     event.events = type | EPOLLIN; // 注册读事件

  118.     int iRet = epoll_ctl(m_epfd, NULL == pHander ? EPOLL_CTL_ADD
  119.      : EPOLL_CTL_MOD, pBase->GetFD(), &event);
  120.     if (iRet == -1)
  121.     {
  122.         std::cout << "epoll_ctl error: " << strerror(errno) << std::endl;
  123.         return -1;
  124.     }

  126.     m_handlerSet.AddFDEventHandler(pBase, EPOLLIN);
  127.     return 0;
  128. }
  129. // 注册写事件
  130. // @return: 0, success. else failed.
  131. int LTReactor::RegisterWriteEvent(FDEventHanderBase* pBase) {
  132.     int type = 0;
  133.     FDEventHanderBase* pHandler = m_handlerSet.GetFDEventHandler(
  134.      pBase->GetFD(), type);

  136.     epoll_event event;
  137.     event.data.fd = pBase->GetFD();
  138.     event.events = type | EPOLLOUT; //注册写事件

  140.     int iRet = epoll_ctl(m_epfd, NULL == pHandler ? EPOLL_CTL_ADD
  141.      : EPOLL_CTL_MOD, pBase->GetFD(), &event);
  142.     if (iRet == -1)
  143.     {
  144.         std::cout << "epoll_ctl error: " << strerror(errno) << std::endl;
  145.         return -1;
  146.     }

  148.     m_handlerSet.AddFDEventHandler(pBase, EPOLLOUT);
  149.     return 0;
  150. }
  151. // 注销读事件
  152. // @return: 0, success. else failed.
  153. int LTReactor::UnRegisterReadEvent(FDEventHanderBase* pBase) {
  154.     int type = 0;
  155.     FDEventHanderBase* pHandler = m_handlerSet.GetFDEventHandler(
  156.      pBase->GetFD(), type);
  157.     epoll_event event;
  158.     event.data.fd = pBase->GetFD();
  159.     event.events = (type & ~EPOLLIN); // 取消读事件

  161.     int iRet = epoll_ctl(m_epfd, NULL == pHandler ? EPOLL_CTL_DEL
  162.      : EPOLL_CTL_MOD, pBase->GetFD(), &event);
  163.     if (iRet == -1)
  164.     {
  165.         std::cout << "epoll_ctl error: " << strerror(errno) << std::endl;
  166.         return -1;
  167.     }

  169.     m_handlerSet.DelFDEventHandler(pBase->GetFD(), EPOLLIN);
  170.     return 0;
  171. }

  173. // 注销写事件
  174. // @return: 0, success. else failed.
  175. int LTReactor::UnRegisterWriteEvent(FDEventHanderBase* pBase) {
  176.     int type = 0;
  177.     FDEventHanderBase* pHandler = m_handlerSet.GetFDEventHandler(
  178.      pBase->GetFD(), type);
  179.     epoll_event event;
  180.     event.data.fd = pBase->GetFD();
  181.     event.events = (type & ~EPOLLOUT); // 取消写事件

  183.     int iRet = epoll_ctl(m_epfd, NULL == pHandler ? EPOLL_CTL_DEL
  184.      : EPOLL_CTL_MOD, pBase->GetFD(), &event);
  185.     if (iRet == -1)
  186.     {
  187.         std::cout << "epoll_ctl error: " << strerror(errno) << std::endl;
  188.         return -1;
  189.     }

  191.     m_handlerSet.DelFDEventHandler(pBase->GetFD(), EPOLLOUT);
  192.     return 0;
  193. }
  194. // 注销读写事件
  195. int LTReactor::UnRegisterAllEvent(FDEventHanderBase* pBase) {
  196.     epoll_event event;
  197.     event.data.fd = pBase->GetFD();
  198.     event.events = EPOLLOUT | EPOLLIN; // 取消读写事件
  199.     int iRet = epoll_ctl(m_epfd, EPOLL_CTL_DEL, pBase->GetFD(), &event);
  200.     if (iRet == -1)
  201.     {
  202.         std::cout << "epoll_ctl error: " << strerror(errno) << std::endl;
  203.         return -1;
  204.     }

  206.     m_handlerSet.DelFDEventHandler(pBase->GetFD(), EPOLLOUT | EPOLLIN);
  207.     return 0;
  208. }
  209. // 注册定时器
  210. int LTReactor::RegisterTimer(TimerEventHanderBase* pBase) {
  211.     m_handlerSet.AddTimerEventHandler(pBase);
  212.     return 0;
  213. }
  214. // 注销定时器
  215. int LTReactor::UnRegisterTimer(TimerEventHanderBase* pBase) {
  216.     m_handlerSet.DelTimerEventHandler(pBase);
  217.     return 0;
  218. }
  219. // 扫描，看是否有定时器的时间到了
  220. void LTReactor::ScanTimer() {
  221.     m_handlerSet.ScanTimer();
  222. }

  224. /* **********************************
  225.  *             处理FD类的基类
  226.  * 当该fd被触发时，调用处理函数来进行处理
  227.  * ***********************************/
  228. // 注册读事件
  229. int FDEventHanderBase::RegisterReadEvent() {
  230.     return m_pReactor->RegisterReadEvent(this);
  231. }
  232. // 注册写事件
  233. int FDEventHanderBase::RegisterWriteEvent() {
  234.     return m_pReactor->RegisterWriteEvent(this);
  235. }
  236. // 注销读事件
  237. int FDEventHanderBase::UnRegisterReadEvent() {
  238.     return m_pReactor->UnRegisterReadEvent(this);
  239. }
  240. // 注销写事件
  241. int FDEventHanderBase::UnRegisterWriteEvent() {
  242.     return m_pReactor->UnRegisterWriteEvent(this);
  243. }

  245. int FDEventHanderBase::GetFD() const {
  246.     return m_fd;
  247. }
  248. void FDEventHanderBase::SetFD(int fd) {
  249.     m_fd = fd;
  250. }
  251. // 设置非阻塞，返回0，success.
  252. int FDEventHanderBase::SetNonBlock(int fd, bool bNonBlock/* = true*/) {
  253.     int flags = fcntl(fd, F_GETFL, 0);
  254.     if (flags == -1)
  255.     {
  256.         return -1;
  257.     }
  258.     if (bNonBlock)
  259.         flags |= O_NONBLOCK;
  260.     else
  261.         flags &= ~O_NONBLOCK;

  263.     if (fcntl(fd, F_SETFL, flags) == -1)
  264.     {
  265.         return -1;
  266.     }

  268.     return 0;
  269. }
  270. /* **********************************
  271.  *        处理timer类的基类
  272.  *    当timer被触发时，调用该类的函数
  273.  * **********************************/
  274. // 注册定时器，@msec，多长时间响一次,单位:毫秒
  275. // 由于epoll精度的限制，这个地方，精度只能达到 毫秒，但不是准确
  276. int TimerEventHanderBase::RegisterTimer(unsigned int msec /*ms*/, bool restart)
  277. {
  278.     m_bRestart = restart;

  280.     m_interval.tv_sec = msec / 1000;
  281.     m_interval.tv_usec = msec % 1000;
  282.     timeval now;
  283.     gettimeofday(&now, NULL);
  284.     timeradd(&now, &m_interval, &m_endtime);
  285.     return m_pReactor->RegisterTimer(this);
  286. }
  287. int TimerEventHanderBase::UnRegisterTimer() {
  288.     return m_pReactor->UnRegisterTimer(this);
  289. }
  290. int TimerEventHanderBase::RegisterTimerAgain() {
  291.     if (timerisset(&m_endtime) && timerisset(&m_interval))
  292.     {
  293.         timeval now;
  294.         gettimeofday(&now, NULL);
  295.         timeradd(&now, &m_interval, &m_endtime);
  296.         return m_pReactor->RegisterTimer(this);
  297.     }
  298.     return -1;
  299. }
  300. /* **********************************
  301.  *             处理类的集合
  302.  *         有个对应关系 -- fd -- 处理函数 -- read/write.
  303.  * *********************************/
  304. // 增加fd事件, @type : EPOLLIN, EPOLLOUT，
  305. void EventHanderSet::AddFDEventHandler(FDEventHanderBase* pHandler, int type) {
  306.     const int fd = pHandler->GetFD();
  307.     std::map<int, FDHandler>::iterator it = m_fdmap.find(fd);
  308.     if (it != m_fdmap.end())
  309.     {
  310.         it->second.type |= type;
  311.         it->second.pHandler = pHandler;
  312.     }
  313.     else
  314.     {
  315.         FDHandler th;
  316.         th.type = type;
  317.         th.pHandler = pHandler;
  318.         m_fdmap.insert(std::make_pair(fd, th));
  319.     }
  320. }
  321. // @type : EPOLLIN, EPOLLOUT
  322. void EventHanderSet::DelFDEventHandler(int fd, int type) {
  323.     std::map<int, FDHandler>::iterator it = m_fdmap.find(fd);
  324.     if (it != m_fdmap.end())
  325.     {
  326.         it->second.type &= ~type;//去掉该属性
  327.         if (0 == it->second.type) //若所有的属性都去掉了，从map中删除该handler.
  328.         {
  329.             m_fdmap.erase(it);
  330.         }
  331.     }
  332. }
  333. // 根据fd，找到 FDEventHanderBase，及type.
  334. FDEventHanderBase* EventHanderSet::GetFDEventHandler(int fd, int& type) {
  335.     std::map<int, FDHandler>::iterator it = m_fdmap.find(fd);
  336.     if (it != m_fdmap.end())
  337.     {
  338.         type = it->second.type;
  339.         return it->second.pHandler;
  340.     }

  342.     return NULL;
  343. }

  345. void EventHanderSet::AddTimerEventHandler(TimerEventHanderBase* pHandler) {
  346.     timeval endtime = pHandler->GetEndTime();
  347.     std::pair<long int, long int> t(endtime.tv_sec, endtime.tv_usec);
  348.     printf("addTimer: [%u:%u] \n", (unsigned int) t.first,
  349.      (unsigned int) t.second);
  350.     m_timerMinHeap.Insert(t);
  351.     m_timerMultiMap.insert(std::make_pair(t, pHandler));
  352. }
  353. void EventHanderSet::DelTimerEventHandler(TimerEventHanderBase* pHandler) {
  354.     timeval endtime = pHandler->GetEndTime();
  355.     std::pair<long int, long int> t(endtime.tv_sec, endtime.tv_usec);
  356.     std::multimap<std::pair<long int, long int>, TimerEventHanderBase*>::iterator
  357.      beg = m_timerMultiMap.lower_bound(t);
  358.     std::multimap<std::pair<long int, long int>, TimerEventHanderBase*>::iterator
  359.      end = m_timerMultiMap.upper_bound(t);
  360.     for (; beg != end; ++beg)
  361.     {
  362.         if (beg->second == pHandler)
  363.         {
  364.             m_timerMultiMap.erase(beg);
  365.             break;
  366.         }
  367.     }
  368. }
  369. void EventHanderSet::ScanTimer() {

  371.     while (true)
  372.     {
  373.         //        printf("m_timerMinHeap: m_max_size[%d] m_cur_size[%d] \n",
  374.         //         m_timerMinHeap.GetMaxSize(), m_timerMinHeap.GetCurSize());
  375.         if (!m_timerMinHeap.IsEmpty())
  376.         {
  377.             std::pair<long int, long int> minTime = m_timerMinHeap.GetMin(); //获取最小时间
  378.             timeval minTimeVal;
  379.             minTimeVal.tv_sec = minTime.first;
  380.             minTimeVal.tv_usec = minTime.second;

  382.             timeval now =
  383.             { 0, 0 };
  384.             gettimeofday(&now, NULL);
  385.             if (timercmp(&now, &minTimeVal, >=)) //当前时间>=触发时间
  386.             {
  387.                 m_timerMinHeap.RemoveMin(); //去掉最小堆里面的时间
  388.                 std::multimap<std::pair<long int, long int>,
  389.                  TimerEventHanderBase*>::iterator beg =
  390.                  m_timerMultiMap.lower_bound(minTime);

  392.                 for (; beg != m_timerMultiMap.upper_bound(minTime);)
  393.                 {
  394.                     TimerEventHanderBase* pHandler = beg->second;
  395.                     pHandler->OnTimeOut();
  396.                     if (pHandler->IsRestart()) //需要重启
  397.                     {
  398.                         AddTimerEventHandler(pHandler);
  399.                     }
  400.                     m_timerMultiMap.erase(beg++);//去掉multimap中的句柄
  401.                 }
  402.             }
  403.             else
  404.             {
  405.                 break;
  406.             }
  407.         }
  408.         else
  409.         {
  410.             break;
  411.         }
  412.     }
  413. }

7. 测试代码

测试代码

  1. #include "reactor.h"
  2. #include <stdio.h>
  3. #include <string.h>
  4. #include <unistd.h>
  5. #include <sys/types.h>
  6. #include <sys/stat.h>
  7. #include <fcntl.h>

  9. class TimerEventHandler_test: public TimerEventHanderBase
  10. {
  11. public:
  12.     TimerEventHandler_test(Reactor * pReactor) :
  13.         TimerEventHanderBase(pReactor)
  14.     {

  16.     }
  17.     void OnTimeOut() {
  18.         timeval t;
  19.         memset(&t, 0, sizeof(t));
  20.         gettimeofday(&t, NULL);
  21.         printf("timeOut: [%u:%u] \n", (unsigned int)t.tv_sec, (unsigned int)t.tv_usec);
  22.     }
  23. };

  25. void reactor_test_1() {
  26.     Reactor * pReactor = new LTReactor;
  27.     pReactor->Init();
  28.     TimerEventHanderBase* pHandler1 = new TimerEventHandler_test(pReactor);
  29.     pHandler1->RegisterTimer(2000, false);
  30.     TimerEventHanderBase* pHandler2 = new TimerEventHandler_test(pReactor);
  31.     pHandler2->RegisterTimer(4000, false);
  32.     TimerEventHanderBase* pHandler3 = new TimerEventHandler_test(pReactor);
  33.     pHandler3->RegisterTimer(8000, false);
  34.     TimerEventHanderBase* pHandler4 = new TimerEventHandler_test(pReactor);
  35.     pHandler4->RegisterTimer(10000, false);
  36.     pReactor->Run();
  37. }
  38. ////////////////////////////////////////////////////////////////////////////
  39. class FileFDEventHander: public FDEventHanderBase
  40. {
  41. public:
  42.     FileFDEventHander(int fd, Reactor* pReactor) :
  43.         FDEventHanderBase(fd, pReactor)
  44.     {

  46.     }

  48.     // 读被触发时，调用该函数
  49.     virtual void OnFDRead() {
  50.         char buf[1024] = {0};
  51.         read(m_fd, buf, sizeof(buf));
  52.         printf("FDReadResult: [%s] \n", buf);
  53.         UnRegisterReadEvent();
  54.     }
  55.     // 写被触发时，调用该函数
  56.     virtual void OnFDWrite() {
  57.         char buf[] = "this is a test file !";
  58.         write(m_fd, buf, strlen(buf));

  60.         UnRegisterWriteEvent();
  61.     }
  62.     // 关闭fd
  63.     virtual void Close() {
  64.         close(m_fd);
  65.     }
  66. };

  68. void reactor_test_2()
  69. {
  70.     int fd = -1;
  71.     const char* path = "/home/ll/work/min_heap/test.txt";
  72.     if(0 != mkfifo(path, 0777)) // 创建一个pipe.
  73.     {
  74.         fd = open(path, O_RDWR);
  75.     }

  77.     Reactor * pReactor = new LTReactor;
  78.     pReactor->Init();

  80.     FDEventHanderBase* pHandler = new FileFDEventHander(fd, pReactor);
  81.     // 注册写事件，在命令行 cat pipe_file_name
  82. //    pHandler->RegisterWriteEvent();
  83. //    sleep(5);
  84. //    printf("sleep 5s over \n");
  85.     // 注册读事件，在命令行 echo "dfsdf" > pipe_file_name
  86.     pHandler->RegisterReadEvent();
  87.     sleep(5);
  88.     printf("sleep 5s over \n");
  89.     pReactor->Run();
  90. }
8. 需要注意的信号
        SIGPIPE：Broken pipe: write to pipe with no readers，对socket的两端A和B，若B已经关闭。当A第一次写时，内核会返回RST给应用程序；当A第二次写，内核会返回SIGPIPE信号给应用程序。所以A端需要处理SIGPIPE信号。SIGPIPE默认动作是进程退出，可以忽略该信号，signal(SIGPIPE, SIG_IGN)。
        SIGINT：当用户按ctl+c时，会产生SIGINT信号。可以用来终止进程。signal(SIGINT, proc_exit);
        SIGKILL / SIGSTOP：不能被捕捉或忽略的信号。提供了一种可以杀死进程的方法。signal(SIGKILL/SIGSTOP, proc_exit);
        SIGTERM：Termination signal，由kill命名发出的系统默认终止信号。signal(SIGTERM, proc_exit);
阅读(6753) | 评论(0) | 转发(4) |
5

上一篇：ubuntu下，编译支持LLRP协议的wireshark

下一篇：用epoll实现一个异步处理的模型(2)

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6