AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

UDT中的epoll

2016-05-21 1073
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： epoll 是为处理大量句柄而改进的poll，在UDT中也有支持。UDT使用了内核提供的epoll，主要是epoll_create，epoll_wait，epoll_ctl，UDT定义了CEPollDesc这个结构来管理epoll的描述符和套接字。
epoll 是为处理大量句柄而改进的poll，在UDT中也有支持。UDT使用了内核提供的epoll，主要是epoll_create，epoll_wait，epoll_ctl，UDT定义了CEPollDesc这个结构来管理epoll的描述符和套接字。
struct CEPollDesc
{
   int m_iID;                                // epoll ID
   std::set<UDTSOCKET> m_sUDTSocksOut;       // set of UDT sockets waiting for write events
   std::set<UDTSOCKET> m_sUDTSocksIn;        // set of UDT sockets waiting for read events
   std::set<UDTSOCKET> m_sUDTSocksEx;        // set of UDT sockets waiting for exceptions
 
   int m_iLocalID;                           // local system epoll ID
   std::set<SYSSOCKET> m_sLocals;            // set of local (non-UDT) descriptors
 
   std::set<UDTSOCKET> m_sUDTWrites;         // UDT sockets ready for write
   std::set<UDTSOCKET> m_sUDTReads;          // UDT sockets ready for read
   std::set<UDTSOCKET> m_sUDTExcepts;        // UDT sockets with exceptions (connection broken, etc.)
};
特别要提醒的是，当对端socket连接中断的时候，也是在m_sUDTReads里的
UDT还实现了一个类来进行各项操作，实现的有
create():创建一个epoll，调用了epoll_create
add_usock():添加一个UDT套接字到epoll
add_ssock():添加一个系统套接字到epoll，调用了epoll_ctl
remove_usock():从epoll中移除一个UDT套接字
remove_ssock():从epoll中移除一个系统套接字，调用了epoll_ctl
wait():等待epoll事件或者超时，调用了epoll_wait
release():关闭并释放一个epoll
 
UDT里对epoll的调用是四段式的，比如add_usock这里，调用顺序是epoll_add_usock()->CUDT::epoll_add_usock()->s_UDTUnited.epoll_add_usock()->CEPoll::add_usock
 
int epoll_add_usock(int eid, UDTSOCKET u, const int* events)
{
   return CUDT::epoll_add_usock(eid, u, events);
}
 
int CUDT::epoll_add_usock(const int eid, const UDTSOCKET u, const int* events)
{
   try
   {
      return s_UDTUnited.epoll_add_usock(eid, u, events);
   }
   catch (CUDTException e)
   {
      s_UDTUnited.setError(new CUDTException(e));
      return ERROR;
   }
   catch (...)
   {
      s_UDTUnited.setError(new CUDTException(-1, 0, 0));
      return ERROR;
   }
}
 
int CUDTUnited::epoll_add_usock(const int eid, const UDTSOCKET u, const int* events)
{
   CUDTSocket* s = locate(u);
   int ret = -1;
   if (NULL != s)
   {
      ret = m_EPoll.add_usock(eid, u, events);
      s->m_pUDT->addEPoll(eid);
   }
   else
   {
      throw CUDTException(5, 4);
   }
 
   return ret;
}
 
int CEPoll::add_usock(const int eid, const UDTSOCKET& u, const int* events)
{
   CGuard pg(m_EPollLock);
 
   map<int, CEPollDesc>::iterator p = m_mPolls.find(eid);
   if (p == m_mPolls.end())
      throw CUDTException(5, 13);
 
   if (!events || (*events & UDT_EPOLL_IN))       //UDT_EPOLL_IN 和UDT_EPOLL_OUT、UDT_EPOLL_ERROR分别是0x1, 0x4, 0x8，可以进行&运算
      p->second.m_sUDTSocksIn.insert(u);
   if (!events || (*events & UDT_EPOLL_OUT))
      p->second.m_sUDTSocksOut.insert(u);
 
   return 0;
}
 
UDT命名空间提供给应用程序调用接口，可成为API层，API层调用CUDT API，这一层主要做错误处理，捕捉下面两层抛出的错误保存起来交给应用程序使用，CUDT API调用CUDTUnited的实现，如果是UDT SOCKET的socket()，bind()，listen()等，到这层也就结束了，不过epoll要多个第四层，再调用CEPoll的实现。现在来看看CUDTUnited和CEPoll的实现。
CUDTSocket* s = locate(u);
 调用CUDTUnited::locate()，根据SocketID，也就是UDT Socket handle在CUDTUnited的std::map<UDTSOCKET, CUDTSocket*> m_Sockets中找到对应的CUDTSocket结构，如果找不到，抛出错误，找到了就调用CEPoll的add_usock实现，根据传的event参数，将socket导入相应的队列。之后调用s->m_pUDT->addEPoll(eid)
void CUDT::addEPoll(const int eid)
{
   CGuard::enterCS(s_UDTUnited.m_EPoll.m_EPollLock);                      //这种通过类来实现加锁解锁的，我第一次见，还挺方便。
   m_sPollID.insert(eid);
   CGuard::leaveCS(s_UDTUnited.m_EPoll.m_EPollLock);
 
   if (!m_bConnected || m_bBroken || m_bClosing)
      return;
 
   if (((UDT_STREAM == m_iSockType) && (m_pRcvBuffer->getRcvDataSize() > 0)) ||
      ((UDT_DGRAM == m_iSockType) && (m_pRcvBuffer->getRcvMsgNum() > 0)))
   {
      s_UDTUnited.m_EPoll.update_events(m_SocketID, m_sPollID, UDT_EPOLL_IN, true);
   }
   if (m_iSndBufSize > m_pSndBuffer->getCurrBufSize())
   {
      s_UDTUnited.m_EPoll.update_events(m_SocketID, m_sPollID, UDT_EPOLL_OUT, true);
   }
}
这里已经开始更新m_sUDTWrites，m_sUDTReads，通过update_events()，如前所述，update_events()也是CEPoll的成员函数。
 
int CEPoll::update_events(const UDTSOCKET& uid, std::set<int>& eids, int events, bool enable)
{
   CGuard pg(m_EPollLock);
 
   map<int, CEPollDesc>::iterator p;
 
   vector<int> lost;
   for (set<int>::iterator i = eids.begin(); i != eids.end(); ++ i)
   {
      p = m_mPolls.find(*i);
      if (p == m_mPolls.end())
      {
         lost.push_back(*i);
      }
      else
      {
         if ((events & UDT_EPOLL_IN) != 0)
            update_epoll_sets(uid, p->second.m_sUDTSocksIn, p->second.m_sUDTReads, enable);   //UDT_EPOLL_IN 和UDT_EPOLL_OUT、UDT_EPOLL_ERROR分别是0x1, 0x4, 0x8，可以进行&运算
         if ((events & UDT_EPOLL_OUT) != 0)
            update_epoll_sets(uid, p->second.m_sUDTSocksOut, p->second.m_sUDTWrites, enable);
         if ((events & UDT_EPOLL_ERR) != 0)
            update_epoll_sets(uid, p->second.m_sUDTSocksEx, p->second.m_sUDTExcepts, enable);
      }
   }
 
   for (vector<int>::iterator i = lost.begin(); i != lost.end(); ++ i)
      eids.erase(*i);
 
   return 0;
}
 
void update_epoll_sets(const UDTSOCKET& uid, const set<UDTSOCKET>& watch, set<UDTSOCKET>& result, bool enable)
{
   if (enable && (watch.find(uid) != watch.end()))
   {
      result.insert(uid);
   }
   else if (!enable)
   {
      result.erase(uid);
   }
}
 
最后说下wait函数的实现，一样是四段式，就跳过前面三段，直接看第四段
 
int CEPoll::wait(const int eid, set<UDTSOCKET>* readfds, set<UDTSOCKET>* writefds, int64_t msTimeOut, set<SYSSOCKET>* lrfds, set<SYSSOCKET>* lwfds)
{
   // if all fields is NULL and waiting time is infinite, then this would be a deadlock   都空的的话，会死锁，抛出异常
   if (!readfds && !writefds && !lrfds && lwfds && (msTimeOut < 0))
      throw CUDTException(5, 3, 0);
 
   // Clear these sets in case the app forget to do it.   清空
   if (readfds) readfds->clear();
   if (writefds) writefds->clear();
   if (lrfds) lrfds->clear();
   if (lwfds) lwfds->clear();
 
   int total = 0;
 
   int64_t entertime = CTimer::getTime();
   while (true)
   {
      CGuard::enterCS(m_EPollLock);
 
      map<int, CEPollDesc>::iterator p = m_mPolls.find(eid);
      if (p == m_mPolls.end())
      {
         CGuard::leaveCS(m_EPollLock);
         throw CUDTException(5, 13);
      }
 
      if (p->second.m_sUDTSocksIn.empty() && p->second.m_sUDTSocksOut.empty() && p->second.m_sLocals.empty() && (msTimeOut < 0))
      {
         // no socket is being monitored, this may be a deadlock   都空的的话，会死锁，抛出异常
         CGuard::leaveCS(m_EPollLock);
         throw CUDTException(5, 3);
      }
 
      // Sockets with exceptions are returned to both read and write sets.           把m_sUDTReads和m_sUDTExcepts都读到readfds里
      if ((NULL != readfds) && (!p->second.m_sUDTReads.empty() || !p->second.m_sUDTExcepts.empty()))
      {
         *readfds = p->second.m_sUDTReads;
         for (set<UDTSOCKET>::const_iterator i = p->second.m_sUDTExcepts.begin(); i != p->second.m_sUDTExcepts.end(); ++ i)
            readfds->insert(*i);
         total += p->second.m_sUDTReads.size() + p->second.m_sUDTExcepts.size();
      }
      if ((NULL != writefds) && (!p->second.m_sUDTWrites.empty() || !p->second.m_sUDTExcepts.empty()))           //把m_sUDTWrites和m_sUDTExcepts都读到writefds里
      {
         *writefds = p->second.m_sUDTWrites;
         for (set<UDTSOCKET>::const_iterator i = p->second.m_sUDTExcepts.begin(); i != p->second.m_sUDTExcepts.end(); ++ i)
            writefds->insert(*i);
         total += p->second.m_sUDTWrites.size() + p->second.m_sUDTExcepts.size();
      }
 
      if (lrfds || lwfds)     //读系统套接字
      {
         #ifdef LINUX
         const int max_events = p->second.m_sLocals.size();
         epoll_event ev[max_events];
         int nfds = ::epoll_wait(p->second.m_iLocalID, ev, max_events, 0);
 
         for (int i = 0; i < nfds; ++ i)
         {
            if ((NULL != lrfds) && (ev[i].events & EPOLLIN))
           {
               lrfds->insert(ev[i].data.fd);
               ++ total;
            }
            if ((NULL != lwfds) && (ev[i].events & EPOLLOUT))
            {
               lwfds->insert(ev[i].data.fd);
               ++ total;
            }
         }
         #else
         //currently "select" is used for all non-Linux platforms.
         //faster approaches can be applied for specific systems in the future.
 
         //"select" has a limitation on the number of sockets
 
         fd_set readfds;
         fd_set writefds;
         FD_ZERO(&readfds);
         FD_ZERO(&writefds);
 
         for (set<SYSSOCKET>::const_iterator i = p->second.m_sLocals.begin(); i != p->second.m_sLocals.end(); ++ i)
         {
            if (lrfds)
               FD_SET(*i, &readfds);
            if (lwfds)
               FD_SET(*i, &writefds);
         }
 
         timeval tv;
         tv.tv_sec = 0;
         tv.tv_usec = 0;
         if (::select(0, &readfds, &writefds, NULL, &tv) > 0)
         {
            for (set<SYSSOCKET>::const_iterator i = p->second.m_sLocals.begin(); i != p->second.m_sLocals.end(); ++ i)
            {
               if (lrfds && FD_ISSET(*i, &readfds))
               {
                  lrfds->insert(*i);
                  ++ total;
               }
               if (lwfds && FD_ISSET(*i, &writefds))
               {
                  lwfds->insert(*i);
                  ++ total;
               }
            }
         }
         #endif
      }
 
      CGuard::leaveCS(m_EPollLock);
 
      if (total > 0)
         return total;
 
      if ((msTimeOut >= 0) && (int64_t(CTimer::getTime() - entertime) >= msTimeOut * 1000LL))
         throw CUDTException(6, 3, 0);
 
      CTimer::waitForEvent();
   }
 
   return 0;
}
文章标签：
API
关系型数据库
Linux
沈小错
目录
相关文章
菊头蝙蝠
|
6月前
|
存储 网络协议
TCP服务器 IO多路复用的实现：select、poll、epoll
TCP服务器 IO多路复用的实现：select、poll、epoll
菊头蝙蝠
92 0
cheems~
|
6月前
|
API C++
socket编程之常用api介绍与socket、select、poll、epoll高并发服务器模型代码实现(1)
前言   本文旨在学习socket网络编程这一块的内容，epoll是重中之重，后续文章写reactor模型是建立在epoll之上的。
cheems~
80 0
云栖号资讯小哥
|
监控
浅谈select，poll和epoll的区别
云栖号资讯：【点击查看更多行业资讯】在这里您可以找到不同行业的第一手的上云资讯，还在等什么，快来！ select，poll和epoll其实都是操作系统中IO多路复用实现的方法。 select select方法本质其实就是维护了一个文件描述符（fd）数组，以此为基础，实现IO多路复用的功能。
云栖号资讯小哥
3701 0
浅谈select，poll和epoll的区别
monGyrate
|
3月前
|
监控
【网络编程】poll函数
【网络编程】poll函数
monGyrate
30 0
穿过生命散发芬芳
|
6月前
|
监控 Linux
epoll 的用法
【4月更文挑战第16天】epoll 通过改进的接口设计，避免了用户态 - 内核态频繁的数据拷贝，大大提高了系统性能。在使用 epoll 的时候，我们一定要理解条件触发和边缘触发两种模式。
穿过生命散发芬芳
50 0
高二的笔记
|
6月前
|
Linux
epoll
epoll
高二的笔记
43 0
小小工匠
|
6月前
|
Unix Linux API
高性能网络编程 - select、 poll 、epoll 、libevent
高性能网络编程 - select、 poll 、epoll 、libevent
小小工匠
88 0
x64.ink
|
Linux Windows
Epoll和IOCP的比较
x64.ink
12456 0
xumaojun
poll&&epoll实现分析（二）——epoll实现
Epoll实现分析——作者：lvyilong316 通过上一章分析，poll运行效率的两个瓶颈已经找出，现在的问题是怎么改进。首先，如果要监听1000个fd，每次poll都要把1000个fd 拷入内核，太不科学了，内核干嘛不自己保存已经拷入的fd呢？答对了，epoll就是自己保存拷入的fd，它的API就已经说明了这一点——不是 epoll_wait的时候才传入fd，而是通过epoll_ctl把所有fd传入内核再一起"wait"，这就省掉了不必要的重复拷贝。
xumaojun
1038 0
javaedge
epoll和select
epoll和select 假设你在大学读书，住的宿舍楼有很多间房间，你的朋友要来找你。 select版宿管大妈就会带着你的朋友挨个房间去找，直到找到你为止。
javaedge
887 0