关于muduo网络库的注解

http://blog.csdn.net/liuxuejiang158blog/article/details/17056537#comments

  注：muduo用C++实现蛮有意思的，其大量使用boost的shared_ptr,bind,function实现对象生命期控制、事件回调机制，且基于对象编程而非面向对象编程。在此记点笔记吧，以备后查。

文字部分：

1 Reactor模式的实现：关键是三个类：Channel,Poller,EventLoop。

 class Channel：事件分发器，其记录了描述符fd的注册事件和就绪事件，及就绪事件回调比如可读回调readCallback。其和文件描述符fd是一一对应的关系，但其不拥有fd。当一个fd想要注册事件并在事件就绪时执行相应的就绪事件回调时，首先通过Channel::update(this)->EventLoop::updateChannel(Channel*)->Poller::updateChannel(Channel*)调用链向poll系统调用的侦听事件表注册或者修改注册事件。Channel作为是事件分发器其核心结构是Channel::handleEvent()该函数执行fd上就绪事件相应的事件回调，比如fd可读事件执行readCallback()。Channel应该还具有一个功能是：Channel::~Channel()->EventLoop::removeChannel(Channel*)->Poller::removeChannel(Channel*)将Poller中的Channel*移除防止空悬指针。这是因为Channel的生命周期和Poller/EventLoop不一样长。其关键数据成员：int fd_文件描述符，int events_文件描述符的注册事件，int revents_文件描述符的就绪事件，及事件回调readCallback_,writeCallback...

 class Poller：实现IO multiplexing,其功能仅仅是poll系统调用的简单封装，其生命周期和EventLoop一样长，Poller不拥有Channel支持有Channel*指针(因此必要时候需要Channel自己解注册以防空悬指针)。Poller的关键数据成员是:vector<struct pollfd> pollfds_事件结构体数组用于poll的第一个参数;map<int,Channel*> channels_用于文件描述符fd到Channel的映射便于快速查找到相应的Channel,如poll返回后遍历pollfds_找到就绪事件的fd再通过channels_找到fd的channel然后就可以通过Channel::handleEvent()执行就绪事件回调，值得注意的是Channel中已经记录了fd所以fd和Channel完成了双射。关键的成员函数：Poller::poll(int timeoutMs,vector<Channel*> activeChannels)其调用poll侦听事件集合，并在timoutMs时间内就绪的事件集合通过activeChannels返回。这里Poller::poll()在poll返回后本可以执行Channel::handleEvent()将就绪事件的回调执行了，但是并没有这样做的原因是，Channel::handleEvent()可能修改Poller的两个容器,即添加或删除Channel*，在遍历容器的时候修改容器是非常危险的，同时为了简化Poller，Poller的职责仅仅是IO复用，至于事件分发还是交给Channel自己完成。  class EventLoop: 事件循环。先看一个调用链：EventLoop::loop()->Poller::poll()通过此调用链获得一个vector<Channel*> activeChannels_的就绪事件集合，再遍历该容器，执行每个Channel的Channel::handleEvent()完成相应就绪事件回调。至此一个完整的Reactor模式即完成。注意这里的Reactor遵循one loop per thread即所在一个线程中完成，故并没有涉及到线程同步机制，以后可能会有其它线程调用这三个类，则通过线程转移函数将一些操作转移到一个线程中完成。(若一个函数既可能加锁情况下使用有可能在未加锁情况下使用，那么就拆成两个函数，需要加锁的函数去调用不需要加锁的函数。线程转移实现就可以通过两个函数实现，如：假设类one隶属于线程B，线程A调用one的方法fun，fun向one注册一个回调，从而将具体操作转移到one的所属线程B中去执行。) 2 定时器: 主要有几个类,Timer定时器包含超时回调，TimerId定时器加上一个唯一的ID，Timestamp时间戳，TimerQueue管理所有的定时器。传统的定时通过select/poll实现，现在通过timerfd实现定时，采用文件描述符实现定时将有利于统一事件源。这些将为EventLoop实现定时功能。 Timer：定时器，具有一个超时时间和超时回调。超时时间由当前时间戳加上一个超时时间生成一个绝对时间。定时器回调函数timerCallback。 TimerQueue: 定时器队列，用于管理所有的定时器,当定时器超时后执行相应的Timer::run()定时器回调。采用set<pair<TimeStamp,Timer*> >存储所有未超时的定时器，这里采用pair<TimeStamp,Timer*>的原因是在一个时间点可能有多个时间戳TimeStamp超时，而查找只返回一个。通过给timerfd一个超时时间实现超时计时，通过Channel管理timerfd，然后向EventLoop和Poller注册timerfd的可读事件，当timerfd的可读事件就绪时表明一个超时时间点到了，TimerQueue::handleRead()遍历set容器找出那些超时的定时器并执行Timer::run()实现超时回调。timerfd怎么实现多个定时器超时计时的呢？每次向set插入一个定时器Timer的时候就比较set的头元素的超时时间，若新插入的超时时间小，则更新timerfd的时间，从而保证timerfd始终是set中最近的一个超时时间。当timerfd可读时，需要遍历容器set，因为可能此时有多个Timer超时了(尽管tiemrfd是当前最小的定时时间)。为了复用定时器，每次执行完定时器回调后都要检查定时器是否需要再次定时。这里的关键是采用timerfd实现统一事件源。 3 class EventLoop的改动，实现用户定时回调：当有了定时器TimerQueue后，EventLoop就可以实现几个定时器接口：EventLoop::runAt(TimeStamp,TimerCallback)在一个绝对时间执行一个回调TimerCallback;EventLoop::runAfter(double delay,TimerCallback)实现一个相对时间回调，其内部实现是当前时间戳TimeStamp加上delay后形成一个绝对时间戳然后调用EventLoop::runAt(); EventLoop::runEvery(double interval,TimerCallback)实现周期回调，这里将使用到TimerQueue执行完超时回调后会检查定时器是否需要再次定时的功能。 4 class EventLoop的改动，实现用户指定任务回调：EventLoop::runInLoop(boost::function<void()>)，若是EventLoop隶属的线程调用EventLoop::runInLoop()则EventLoop马上执行;若是其它线程调用则执行EventLoop::queueInLoop(boost::function<void()>将任务添加到队列中(这里就是前面说的线程转移)。EventLoop如何获得有任务这一事实呢？通过eventfd可以实现线程间通信，具体做法是：其它线程向EventLoop::vector<boost::function<void()> >添加任务T，然后通过EventLoop::wakeup()向eventfd写一个int，eventfd的回调函数EventLoop::handleRead()读取这个int，从而相当于EventLoop被唤醒，此时loop中遍历队列执行堆积的任务。这里采用Channel管理eventfd，Poller侦听eventfd体现了eventfd可以统一事件源的优势。 5 实现线程安全的class TimerQueue：原来添加定时器的调用链：TimerQueue::addTimer()->TimerQueue::insert()实现添加的，但是这只能实现TimerQueue所属的线程执行，若其它线程想向此IO线程添加一个定时任务是不安全的。          为了实现添加定时器Timer到set的线程安全性，将定时器添加函数TimerQueue::addTimer()分拆为两部分：TimerQueue::addTimer()只负责转发，addTimerInLoop()实现具体的定时器添加。具体的调用链为：TimerQueue::addTimer()->EventLoop::runInLoop(TimerQueue::runInLoop)->TimerQueue::runInLoop()->TimerQueue::insert()，可以看出通过把TimerQueue::runInLoop()这个回调任务添加EventLoop::runInLoop()中从而实现将添加定时器这一操作转移到IO线程来做。TimerQueue::insert()插入一个新的定时器后将检查当前最近的超时时间，若最近的超时时间变了则重置timerfd的计时。 6 class  EventLoopThread: 启动一个线程执行一个EventLoop,其语义和"one loop per thread“相吻合。注意这里用到了互斥量和条件变量,这是因为线程A创建一个EventLoopThread对象后一个运行EventLoop的线程已经开始创建了，可以通过EventLoopThread::startLoop()获取这个EventLoop对象，但是若EventLoop线程还没有创建好，则会出错。所以在创建EventLoop完成后会执行condititon.notify()通知线程A，线程A调用EventLoopThread::startLoop()时调用condition.wai()等待，从而保证获取一个创建完成的EventLoop.毕竟线程A创建的EventLoop线程，A可能还会调用EventLoop执行一些任务回调呢。 7 class Acceptor： 用于accept一个TCP连接，accept接受成功后通知TCP连接的使用者。Acceptor主要是供TcpServer使用的，其生命期由后者控制。一个Acceptor相当于持有服务端的一个socket描述符，该socket可以accept多个TCP客户连接，这个accept操作就是Acceptor实现的。            这里封装了一些常用的网络相关的数据结构和操作，如class InetAddress表示sockaddr_in的封装，如可以通过ip地址和port端口生成一个sockaddr_in; class Socket封装了部分关于socket套接字的操作，如Socket::bindAddress(InetAddress&)将socket和一个sockaddr_in地址绑定，Socket::accept(InetAddress& peerAddr)将一个socket允许连接一个客户端地址peerAddr，Socket::listen()监听socket，Socket::shutdownWrite()实现关闭socket的写。               Acceptor在构造的时候会创建一个socket描述符acceptSocket_(这是一个Socket类型即socket的RAII封装)，并通过一个Channel(注册事件及回调函数)管理acceptSocket_::fd成员(即socket描述符)，一旦该socket可读即有TCP客户连接请求，则Channel::handleEvent()将会调用Acceptor::hanleRead()执行accept一个TCP客户连接。Acceptor::handleRead()还会将新的TCP客户连接和客户端地址通过回调函数newConnectionCallback(connfd,peerAddr)传给该TCP客户连接的使用者，通常是TcpServer类，这里的回调函数newConnectionCallback是在Acceptor::setNewConnectionCallback(newConnectionCallback)指定的。值得注意的是这里又是统一事件源的思想，即通过Channel和Poller管理事件。Acceptor::listen()的工作是：启动acceptSocket_::listen()监听socket描述符，并通过Channel::enableReading()将socket的可读事件注册到Poller的事件集合中。 8 class TcpServer: 管理所有的TCP客户连接，TcpServer供用户直接使用，生命期由用户直接控制。用户只需设置好相应的回调函数(如消息处理messageCallback)然后TcpServer::start()即可。             这里先假设每个TCP客户连接由一个类TcpConenction管理(具体执行消息的接收发送之类的)，而TcpServer的工作就是管理这些TcpConenction，TcpConnection将在后面给出。假设TcpServer持有boost::scoped_ptr<TcpConnection>的指针TcpConnectionPtr。               TcpServer在构造时接收一个由IP地址和port构成的InetAddress参数，并将此地址传给Acceptor用于接收该地址的TCP连接请求。              TcpServer持有scoped_ptr<Acceptor> acceptor_用于接收TcpServer监听端口上的TCP连接请求，注意Accpetor每次accept连接后都要将新连接的描述符connfd和地址peerAddr返回给使用者，这里TcpServer在构造时通过accptor_->setNewConnectionCallback(bind(&TcpServer::newConnection,this,_1,_2))将TcpServer::newConnection传给Acceptor，acceptor_在接受TCP客户连接后将调用TcpServer::newConnection(connfd,peerAddr)，而TcpSrever::newConnection()的主要功能就是为<connfd,peerAddr>创建一个TcpConnection管理该TCP客户连接，并向TcpConnection注册一些回调函数，比如:connectionCallback主要是在TcpServer中由用户指定的一些连接处理函数最后一路经由TcpSrever传到TcpConnection中才被调用，此外还有用户指定的消息处理回调等都是经由TcpServer传给TcpConnection中去具体执行。此外TcpServer::newConnection()中还会执行TcpConnection::connectEstablished()该函数将会使这个具体的TcpConnection连接对应的描述符connfd加入poll的事件表，即也是通过一个Channel管理一个具体的TCP客户连接。用户向TcpServer注册连接回调函数的调用链：用户在创建TcpServer后TcpServer::setConnectionCallback()接收用户注册的连接回调函数;同时在TcpServer创建时会向Acceptor注册回调：TcpServer::TcpServer()->Acceptor::setNewConnecioncallback()后有新TCP连接Acceptor接受连接，并执行回调给连接使用者：Acceptor::handelRead()->newConnection()/Tcpserver::newConnection()->TcpConnection::connectEstablished()/并向TcpConnection注册用户注册的Callback函数。               TcpServer采用map<string,TcpConnectionPtr>管理所有的TCP客户连接，其中string是由TcpServer的服务端地址加上一个int构成表示TcpConnectionPtr的名字。               TcpServer中由用户指定的回调有：connectionCallback当TcpConenction建立时调用(由TcpConnection::connectEstablished()调用connectionCallback())用于执行用户指定的连接回调。messageCallback当TcpConenction有网络消息的时候执行该函数由Channel::handleEvent()->TcpConnection::handleRead()->messageCallback()。writeCompleteCallback由用户指定的当TCP连接上的消息发送完毕时执行的回调。这些函数都是用户在TcpServer创建后通过TcpServer::set*Callback系列函数注册的。当Acceptor接受一个新的TCP连接时执行Acceptor::handleRead()->TcpServer::newConnection()->TcpConnection::set*Callback()这样完成用于指定函数的传递。那么执行呢？这个要在TcpConenction对应的socket事件就绪时可读/可写时由Channel::handEvent()执行这些用户指定的回调。               TcpServer::removeConnection()主要功能从TcpServer中移除一个TcpConnection，但是不能直接移除，而要通过线程转移函数完成。TcpServer::removeConenction()将执行EventLoop::runInLoop(bind(&TcpServer::removeConnectionInLoop)->EventLoop::runInLoop()->TcpServer::removeConnectionInLoop()                TcpServer::removeConenctionInLoop()将一个TcpConnection从TcpServer中移除，并向EventLoop注册回调EventLoop::runInLoop(bind(&TcpConenction::connectDestroyed)),然后执行TcpConnection::connectDestroyed()。 9 class TcpConnection: 用于管理一个具体的TCP客户连接，比如消息的接收与发送，完成用户指定的连接回调connectionCallback。这里采用shared_ptr管理TcpConnection，因此其public继承boost::enable_shared_from_this<TcpConnection>。               TcpConnection构造时接收参数有TCP连接的描述符sockfd，服务端地址localAddr,客户端地址peerAddr，并通过Socket封装sockfd。且采用Channel管理该sockfd，向Channel注册TcpConection的可读/可写/关闭/出错系列回调函数，用于Poller返回就绪事件后Channel::handleEvent()执行相应事件的回调。               TcpConnection有四个状态：kConnecting正在连接，kConnected已连接，kDisconnecting正在断开，kDisconnected已断开。               TcpConnection:有一些列函数用于TcpServer为连接指定事件回调函数，如TcpConnection::setConnectionCallback/setCloseback等是在TcpServer::newConnection()中注册的回调函数，并且当Acceptor接受一个新TCP连接后执行回调TcpServer::newConnection()，该回调创建一个TcpConenction对象并将用户指定的回调函数通过TcpConnection::set*Callback函数传给TcpConneciton。              TcpConnection有一些列函数用户处理sockfd上的事件回调函数，如TcpConnection::handleRead()是在Poller返回sockfd可读事件时由Channel::handleEvent()调用的。类似还有TcpConnection::handleWrite()等。              TcpConnection::send(string& message)->EventLoop::runInLoop(bind(&TcpConnection::sendInLoop(string& message))->EventLoop::doPendingFunctors()->TcpConnection::sendInLoop(string& message)保证消息发送的线程安全，后者通过write系统调用发送消息。              TcpConnection::shutdown()->EventLoop::runInLoop(bind(&TcpConnection::shutdownInLoop())->EventLoop::doPendingFunctors()->TcpConnection::shutdownInLoop()类似上面，通过线程转移操作实现安全关闭TCP连接。               TcpConnection中增加了用户指定系列回调函数conenctionCallback.messageCallback,writeCompleteCallback这些都是用户通过TcpServer传给TcpConnection，在TcpServer中已经描述过了。当Poller返回TcpConenction对应的Socket就绪事件时Channel::handleEvent()->TcpConnection::handle些列函数->执行这一些列回调。                TcpConnection::closeCallback()不是给用户使用的，而是通知TcpServer或TcpClient移除它们容器中的TcpConnectionPtr。该函数如何设定的呢？当Acceptor接受一个TCP连接时:Channel::handelEvent()->Acceptor::handleRead()->TcpServer::newConenction()中新建一个TcpConnection并通过TcpConnection::setCloseCallback(bind(&TcpSerer::,removeConenction,this,_1))这样才完成将TcpServer::removeChannel()函数传递给TcpConnection::closeCallback()。closeCallback()何时执行呢？当由Channel管理的TcpConnection对应的Socket发生POLLHUP事件(该事件由Poller返回)就绪时，Channel::handleEvent()->TcpConnection::handleClose()->TcpConnection::closeCallback()->TcpServer::removeConnection()。                 TcpConnection::connectEstablished()连接建立回调函数(不是用户通过TcpServer指定的connectionCallback),该函数主要的功能：调用Channel::enableReading()将TcpConnection对应的Socket注册到Poller的事件表，执行用户指定的connectionCallback，并将TcpConnection状态置为kConnected。该函数何时被执行呢？回忆前面的Acceptor持有Tcpserver对应的服务端侦听描述符listenfd(由Channel管理)，当listenfd可读表明有TCP连接请求，此时Channel::handleEvent()->Acceptor::handleRead()->TcpServer::newConnection()->EventLoop::runInLoop(bind(&TcpConnection::connectEstablished))->EventLoop::queueInLoop()->EventLoop::loop()->EventLoop::doPendingFunctors()->TcpConnection::connectEstablished()。可见TcpServer也是通过向EventLoop::runInLoop添加Tcpconnection::conectEsatablished回调，表明TcpServer可TcpConencion可能不再同一个线程，需要通过线程转移来实现调用。                 TcpConenction::connectDestroyed()是TcpConenction析构前调用的最后一个函数，用于通知用户连接已断开。其只要功能是：将TcpConenction状态设为kDisconnected;Channel::disableAll()解除TcpConnection的事件注册，EventLoop::removeChannel()移除该管理TcpConnection的Channel;执行用于指定的回调conenctionCallback。该函数如何调用的呢？这要追溯到TcpServer::newConnection()将TcpServer::removeConenction注册到TcpConnection::closeCallback中，当TcpConnection对应的Socket的POLLHUP事件触发时执行TcpConenction::handleClose()->closeCallback()/TcpServer::removeConenction()->EvetnLoop::runInLoop()->TcpServer::removeInLoop()->EventLoop::runInLoop(bind(&TcpConnection::connectDestroyed))->TcpConnection::connectDestroyed()。                TcpConnection::shutdown()用使用者执行关闭TcpConenction，TcpConnection::shutdown()->EventLoop::runInLoop()->TcpConnection::shutdownInLoop()->socket::shutdown(sockfd,SHUT_WR)//SHUT_WR表示以后的发送将被丢弃。                TcpConnection::handleRead()被Channel::handleEvent()可读事件调用，它的主要工作是通过readv()将socket上的数据读取到Buffer中，并执行用于指定的消息回调messageCallback()。                TcpConnection::handleWrite():被Channel::handleEvent()的可写事件调用，通过write()将Buffer的数据发送出去，若Buffer的数据一次性发送完毕，则执行用户指定的回调writeCompleteCallback()，若一次没有发送完毕，则poll和epoll的LT模式会反复触发可写事件的，所以下次还有机会发送剩余数据。                TcpConnection::handleClose()主要执行Channel::disableAll()和closeCallback()。                TcpConnection有连个Buffer(后面会提到)，inputBuffer_,outputBuffer_管理TcpConenction上的数据接收与发送。inputBuffer由TcpConnection::handleRead()调用(Buffer通过readv集中从fd集中读到内存中)。outputBuffer由TcpConnection::handleWrite()通过write()发送到fd上。                TcpConnection::send()->EventLoop::runInLoop()->TcpConenction::runInLoop()，send()可以用户或者其它线程调用，用于发送消息message。这个函数需要先执行线程安全转移到TcpConenction所在的IO线程执行runInLoop()。runInLoop()函数的功能：首先检查TcpConneciton对应的Socket是否注册了可写事件，若注册了可写事件表明outputBuffer_中已经有数据等待发送，为了保证顺序这次的数据只好outputBuffer_.appen()到Buffer中通过Poller返回POLLOUT事件时Channel::handleEvent()->TcpConenction::handleWrite()来发送outputBuffer_的堆积数据。如果Channel::isWriting()返回false则表明此Socket没有向Poller注册POLLOUT事件也就此前没有数据堆积在outputBuffer_中，此次的消息message可以直接通过write发送，但是如果write没有一次性发送完毕，那么message剩余的数据仍要outputBuffer_::append()到Buffer中，并向Poller注册此Socket的POLLOUT事件，以通过TcpConnection::handleWrite()来发送outputBuffer_的堆积数据。无论是sendInLoop()->write还是Channel::handleEvent()->handleWrite()，只要确定发送完message或者outputBuffer_中的数据，那么都要调用用户指定的回调writeCompleteCallback()。                  此外TcpConenction还需要忽略SIGPIPE信号，做法是在TcpConenction.cc中定义一个IngoreSigPipe类构造时signal(SIGPIPE,SIG_IGN)。                 TcpConnection::setTcpNoDelay()->socketopt(..,TCP_NODELAY..)来关闭Nagle算法。 10  到了这步，EventLoop和Poller都应该具备成员函数removeChannel()用于移除那些管理TcpConenction等的Channel。这里Poller有个成员vector<struct pollfd> pollfds_当移除Channel时有个trick值得注意：先将要移除的Channel::fd()的pollfd和pollfds_的最后一个元素交换swap()然后再调用pollfds_.pop_back可以避免vector删除时元素的移动。这样删除操作的复杂度为O(1)。 11   class Buffer应用层缓冲区：在non-blocking+IO multiplexing中应用层缓冲区是必须的。例如：TcpConection向发送100Kb数据，但是write只发送80Kb，剩下20Kb肯定要阻塞线程等待发送了，这和non-blocking矛盾，因此需要设计Buffer充当中间件置于应用层和底层write/read间从而实现non-blocking。这样，应用层只管生成或者读取数据，至于怎样存取数据和数据量则由Buffer完成。Buffer底层是vector<char>,有一个readerIndex和writerIndex分别表示可读位置和可写位置，这两个位置内的区间表示Buffer已有的数据。值得注意的是Buffer的两个trick：每次读取fd上的数据时通过readv一部分读取到Buffer中，一部分读取到找空间char extrabuf[65535]中，若Buffer装满了即extrabuf中有数据，则需要extrabuf中的数据append到Buffer中，这样做可以在初始时每个连接的Buffer的避免过大造成内存浪费，也避免反复调用read的系统开销，每次Buffer不够时通过extrabuf再append到Buffer中使Buffer慢慢变大;还有一个就是Buffer提供了一个前向空间，在消息序列化完毕时可以通过prepend()将消息的大小添加的Buffer的头部。另外Buffer的readerIndex和writerIndex都是移动的，只要在Buffer的空闲空间不够时才加大Buffer的vector，否则可以通过内部腾挪方式即让readerIndex和writerIndex向前移动(数据也跟着移动)这样Buffer就不增大vector.size(). 12 class TcpServe的改进:  TcpServer有自己的一个EventLoop用来接收新的TCP客户连接，然后从event loop pool中选一个loop给TCP客户连接(即TcpConnection)。这就需要使用class EventLoopThreadPool来创建多个线程每个线程一个EventLoop(one loop per thread)。               TcpServer::newConnection()在创建一个TcpConnection后从EventLoopTreadPool中选一个EventLoop给这个TcpConnection。

              TcpServer::removeConnection()需要拆分成两个函数，因为现在TcpServer和TcpConenction可能不再同一个线程里了，需要通过线程转移函数将移除操作转移到TcpConneciton的IO线程中去。TcpServer::removeConnection()->EventLoop::runInLoop()->TcpServer::removeConnectionInLoop()->EventLoop::runInLoop()->TcpConnection::connectDestroyed()。Tcpserver中erase掉这个TcpConnectionPtr。

13 class Connector:用于发起连接，当socket变得可写时表示连接建立完毕，其间需要处理各种类型的错误。connect返回EAGAIN是真的错误表示暂时没有端口可用，要关闭socket稍后再试;EINPROGRESS是“正在连接”，即使socket可写，也需要用getsockopt(sokfd,SOL_SOCKET,SO_ERROR...)再次确认。超时重连的时间应逐渐延长，需要处理自连接的情况。Connector只负责建立连接，不负责建立TcpConnection，它有一个建立连接回调函数由用户指定(Connector基本是TCP客户端使用，且一个客户端一个Conenctor)。

           Connector有三个状态：kDisconnected未连接，kConnecting正在连接，kConnected已连接。

           Connector构造时指定一个服务端地址InetAddress和一个事件循环EventLoop，状态设为kDisConnected。指定一个最大重试连接时间。

           Connector::start()可以由其它线程调用，该函数内部执行EventLoop::runInLoop(bind(&Connector::startInLoop,this))将通过EventLoop将操作转移到Connector的线程中去。

           Connector::startInLoop()判断此Connector还没有连接，则调用Connector::connect()

           Connector::connect()创建一个sockfd,并调用connect(sockfd,&serverAddress,sizeof(serverAddress))，然后返回一个errno，根据errno的值进行进一步选择是关闭连接，重师连接还是调用已连接函数。

            连接返回errno为EAGAIN则调用Connector::retry()该函数调用EventLoop::runAfter()在一段时间间隔后重试连接。

            errno为EINPROGRESS表示正在连接，则调用Connector::connecting()该函数将该为该连接设置一个Channel来管理连接，并向Channel注册Connector::handleWrite,Conenctor::handleError的回调函数。其中Connector::handleWrite()当正在连接kConnecting则用需要测试该连接(已经可写了还kConencting)可能需要进一步重新连接即调用Connector::retry()注意这时候socketfd需要重新分配了，而Conenctor是可以重复使用的。Cionnector::handleError()执行Connector::removeAndResetChannel()和Connector::retry()重试连接。

            当用户或其它线程调用Connector::start()->EventLoop::runInLoop()->Connector::startInLoop()->Connector::connect()根据errnor执行下面的情形：

EAGAIN：Connector::retry()->EventLoop::runAfter()延迟重连->Conncetor::startInLoop()

EACCESS/EPERM/EBADF:  close(sockfd)

EINPROGRESS：Connector::connecting()该函数向Channel注册回调函数并Channel::enableWriting()关注这个正在连接的socketfd是否可写。

            此后当处于“正在连接kConnecting”的sockfd事件就绪时：Channel::handelEvent()->Connector::handleWrite()/Connector::handleError()

             Connector::handleWrite()若检测Connector状态仍在kConnecting时需要调用getsocketopt(sockfd,SOL_SOCKET,SO_ERROR...)检测，若返回0则执行用户指定的连接回调函数Connector::newConnectionCallback()。

             Connector::handleError()->Conenctor::retry()

14 class TcpClient:每个TcpClinet只管理一个Connector。其内容和TcpServer差不多,TcpClient具备TcpConnection断开连接后重新连接的功能。

             TcpClient构造时new一个Connector，并指定一个EventLoop。用户要指定的connectionCallback,messageCallback回调函数。并设置Connector::setNewConnection(bind(&TcpClient::newConnection,this,_1))将连接建立回调赋给Connector::newConnectionCallback。

             用户调用TcpClient::connect()发起连接->Conenctor::start()->TcpClient::newConnection().

              TcpClinet::newConnection()将new一个TcpConneciton对象conn并设置TcpConnection::setConnecitonCallback/setMessageCallback/setWriteCompleteCallback/setCloseCallback等回调函数。其中setConnectionCallback将用户指定的ConnectionCallback传给TcpConenciton，setMessageCallback将TcpClient中用户指定的messageCallback传给TcpConneciton。TcpConnection::closeCallback是TcpClient::removeConenction。最后TcpClient::newConneciton()将会调用TcpConnection::connectEstablished()。

              TcpClient::removeConnection()，由于TcpClient只管理一个Connector也就是一个TcpConenction它们都在一个线程中，所以不涉及操作线程转移。TcpClient::removeConenction()->EventLoop::queueInLoop()->TcpConnection::connectDestroyed().若有重连的必要将执行Connector::restart()。

14  class Epoller和Poller差不多。

代码部分：

注意这里的代码没有muduo网络库全，挂在这里只是做了些注释，算是留个尸体吧。如有必要请参看muduo。

[cpp] view plain copy

1. #include<iostream>
2. #include<map>
3. #include<string>
4. #include<vector>
5. #include<utility>
6. #include<set>
7. #include<deque>
8. #include<algorithm>
9. #include<boost/any.hpp>
10. #include<boost/enable_shared_from_this.hpp>
11. #include<boost/noncopyable.hpp>
12. #include<boost/scoped_ptr.hpp>
13. #include<boost/shared_ptr.hpp>
14. #include<boost/weak_ptr.hpp>
15. #include<boost/function.hpp>
16. #include<boost/static_assert.hpp>
17. #include<boost/bind.hpp>
18. #include<boost/foreach.hpp>
19. #include<boost/ptr_container/ptr_vector.hpp>
20. #include<errno.h>
21. #include<fcntl.h>
22. #include<stdio.h>
23. #include<strings.h>
24. #include<unistd.h>
25. #include<endian.h>
26. #include<assert.h>
27. #include<stdio.h>
28. #include<stdlib.h>
29. #include<string.h>
30. #include<pthread.h>
31. #include<unistd.h>
32. #include<poll.h>
33. #include<errno.h>
34. #include<signal.h>
35. #include<stdint.h>
36. #include<arpa/inet.h>
37. #include<netinet/tcp.h>
38. #include<netinet/in.h>
39. #include<sys/timerfd.h>
40. #include<sys/syscall.h>
41. #include<sys/time.h>
42. #include<sys/eventfd.h>
43. #include<sys/types.h>
44. #include<sys/socket.h>
45. #include<sys/epoll.h>
46. using namespace std;  
47. using namespace boost;  
48. # define UINTPTR_MAX       (4294967295U)//一个无符号大数
49. /*
50. *互斥量
51. */
52. class Mutex:noncopyable{  
53. public:  
54.         Mutex(){  
55.             pthread_mutex_init(&mutex,NULL);  
56.         }  
57. void lock(){  
58.             pthread_mutex_lock(&mutex);  
59.         }  
60. void unlock(){  
61.             pthread_mutex_unlock(&mutex);  
62.         }  
63.         pthread_mutex_t& get(){  
64. return mutex;  
65.         }  
66. private:  
67.         pthread_mutex_t mutex;  
68. };  
69. /*
70. *互斥量RAII
71. */
72. class MutexLockGuard:noncopyable{  
73. public:  
74. explicit MutexLockGuard(Mutex& mutex):mutex_(mutex){  
75.             mutex_.lock();  
76.         }  
77.         ~MutexLockGuard(){  
78.             mutex_.unlock();  
79.         }  
80. private:  
81.         Mutex& mutex_;  
82. };  
83. /*
84. *条件变量
85. */
86. class Condition:noncopyable{  
87. public:  
88. explicit Condition(Mutex& mutex):mutex_(mutex){  
89.             pthread_cond_init(&pcond_,NULL);  
90.         }  
91.         ~Condition(){  
92.             pthread_cond_destroy(&pcond_);  
93.         }  
94. void wait(){  
95.             pthread_cond_wait(&pcond_,&mutex_.get());  
96.         }  
97. void notify(){  
98.             pthread_cond_signal(&pcond_);  
99.         }  
100. void notifyALL(){  
101.             pthread_cond_broadcast(&pcond_);  
102.         }  
103. private:  
104.         Mutex& mutex_;  
105.         pthread_cond_t pcond_;  
106. };  
107. /*
108. *倒计时闩
109. */
110. class CountDownLatch{  
111. public:  
112.         CountDownLatch(int count):mutex_(),condition_(mutex_),count_(count){}  
113. void wait(){  
114.             MutexLockGuard lock(mutex_);  
115. while(count_>0)  
116.                 condition_.wait();  
117.         }  
118. void countDown(){  
119.             MutexLockGuard lock(mutex_);  
120.             --count_;  
121. if(count_==0)  
122.                 condition_.notifyALL();  
123.         }  
124. private:  
125. mutable Mutex mutex_;  
126.         Condition condition_;  
127. int count_;  
128. };  
129. /*
130.  *线程类Thread
131.  */
132. __thread pid_t t_cacheTid=0;//线程私有数据线程ID避免通过系统调用获得ID
133. class Thread:noncopyable{  
134. public:  
135. typedef function<void()> ThreadFunc;//线程需要执行工作函数
136. explicit Thread(const ThreadFunc& a,const string& name=string()):started_(false),  
137.             joinded_(false),pthreadID_(0),tid_(new pid_t(0)),func_(a),name_(name){  
138.             }  
139.         ~Thread(){  
140. if(started_&&!joinded_){  
141.                 pthread_detach(pthreadID_);//分离线程
142.             }  
143.         }  
144. void start();  
145. /*
146.         {
147.             assert(!started_);
148.             started_=true;
149.             if(pthread_create(&pthreadID_,NULL,&startThread,NULL)){
150.                 started_=false;
151.                 abort();//终止进程刷新缓冲区
152.             }
153.         }
154.         *///###1###使用此处会出错详见http://cboard.cprogramming.com/cplusplus-programming/113981-passing-class-member-function-pthread_create.html  
155. void join(){//等待线程执行完工作函数
156.             assert(started_);  
157.             assert(!joinded_);  
158.             joinded_=true;  
159.             pthread_join(pthreadID_,NULL);  
160.         }  
161.         pid_t tid() const{  
162. if(t_cacheTid==0){//如果没有缓存t_cacheTid则获取线程ID否则直接通过线程私有数据返回ID减少系统调用
163.                 t_cacheTid=syscall(SYS_gettid);  
164.             }  
165. return t_cacheTid;  
166.         }  
167. const string& name() const{  
168. return name_;  
169.         }  
170. //void* startThread(void* arg){//###1###
171. void startThread(){  
172.             func_();  
173.         }  
174. private:  
175. bool started_;  
176. bool joinded_;  
177.         pthread_t pthreadID_;  
178.         shared_ptr<pid_t> tid_;  
179.         ThreadFunc func_;  
180.         string name_;  
181. };  
182. void* threadFun(void* arg){//采用间接层执行工作函数
183.     Thread* thread=static_cast<Thread*>(arg);  
184. thread->startThread();  
185. return NULL;  
186. }  
187. void Thread::start(){  
188.     assert(!started_);  
189.     started_=true;  
190. if(pthread_create(&pthreadID_,NULL,threadFun,this)){  
191.         started_=false;  
192.         abort();//终止进程刷新缓冲区
193.     }  
194. }  
196. /*
197.  * 线程局部数据TSD
198.  */
199. template<typename T>  
200. class ThreadLocal:noncopyable{  
201. public:  
202.         ThreadLocal(){  
203.             pthread_key_create(&pkey_,&destructor);//每个线程会设定自己的pkey_并在pthread_key_delete执行destructor操作
204.         }  
205.         ~ThreadLocal(){  
206.             pthread_key_delete(pkey_);//执行destructor操作
207.         }  
208.         T& value(){//采用单件模式，此处不会跨线程使用故不存在非线程安全的singleton问题
209.             T* perThreadValue=static_cast<T*>(pthread_getspecific(pkey_));  
210. if(!perThreadValue){  
211.                 T* newObj=new T();  
212.                 pthread_setspecific(pkey_,newObj);  
213.                 perThreadValue=newObj;  
214.             }  
215. return *perThreadValue;  
216.         }  
217. private:  
218. static void destructor(void* x){//清除私有数据
219.             T* obj=static_cast<T*>(x);  
220. delete obj;  
221.         }  
222. private:  
223.         pthread_key_t pkey_;  
224. };  
225. /*
226.  * 线程池
227.  */
228. class ThreadPool:noncopyable{  
229. public:  
230. typedef function<void()> Task;//线程工作函数
231. explicit ThreadPool(const string& name=string()):mutex_(),cond_(mutex_),name_(name),running_(false){  
232.         }  
233.         ~ThreadPool(){  
234. if(running_){  
235.                 stop();//等待所有线程池中的线程完成工作
236.             }  
237.         }  
238. void start(int numThreads){  
239.             assert(threads_.empty());  
240.             running_=true;  
241.             threads_.reserve(numThreads);  
242. for(int i=0;i<numThreads;i++){  
243.                 threads_.push_back(new Thread(bind(&ThreadPool::runInThread,this)));//池中线程运行runInThread工作函数
244.                 threads_[i].start();  
245.             }  
246.         }  
247. void stop(){  
248.             running_=false;//可以提醒使用者不要在此后添加任务了，因为停止池但是池还要等待池中线程完成任务
249.             cond_.notifyALL();//唤醒池中所有睡眠的线程
250.             for_each(threads_.begin(),threads_.end(),bind(&Thread::join,_1));//等待池中线程完成
251.         }  
252. void run(const Task& task){  
253. if(running_){//###4###防止停止池运行后还有任务加进来
254. if(threads_.empty()){//池中没有线程
255.                     task();  
256.                 }  
257. else{  
258.                     MutexLockGuard guard(mutex_);//使用RAII mutex保证线程安全
259.                     queue_.push_back(task);  
260.                     cond_.notify();  
261.                 }  
262.             }  
263. else{  
264.                 printf("线程池已停止运行\n");  
265.             }  
266.         }  
267. bool running(){//使用者可以获取线程池的运行状态
268. return running_;  
269.         }  
270. private:  
271. void runInThread(){//线程工作函数
272. while(running_){//###2###
273.                 Task task(take());  
274. if(task){//task可能意外的为NULL
275.                     task();  
276.                 }  
277.             }  
278.         }  
279.         Task take(){  
280.             MutexLockGuard guard(mutex_);  
281. while(queue_.empty()&&running_){//###3###和###2###不能保证在池停止运行但是线程还没有完成操作期间安全。假设此期间有任务添加到池中，且某个线程A执行到###2###后马上被切换了，池running_=false停止运行，A被切换后运行执行###3###处无意义啊，因为池已经停止运行了。所以###4###是有必要提醒使用者池停止这一情景
282.                 cond_.wait();//池中没有任务等待
283.             }  
284.             Task task;  
285. if(!queue_.empty()){  
286.                 task=queue_.front();  
287.                 queue_.pop_front();  
288.             }  
289. return task;  
290.         }  
291.         Mutex mutex_;  
292.         Condition cond_;  
293.         string name_;  
294.         ptr_vector<Thread> threads_;//智能指针容器
295.         deque<Task> queue_;  
296. bool running_;  
297. };  
298. /*
299.  * 原子类型
300.  */
301. template<typename T>  
302. class AtomicIntegerT : boost::noncopyable  
303. {  
304. public:  
305.         AtomicIntegerT()  
306.             : value_(0){}  
307.         T get() const
308.         {  
309. return __sync_val_compare_and_swap(const_cast<volatile T*>(&value_), 0, 0);  
310.         }  
311.         T getAndAdd(T x)  
312.         {  
313. return __sync_fetch_and_add(&value_, x);  
314.         }  
315.         T addAndGet(T x)  
316.         {  
317. return getAndAdd(x) + x;  
318.         }  
319.         T incrementAndGet()  
320.         {  
321. return addAndGet(1);  
322.         }  
323. void add(T x)  
324.         {  
325.             getAndAdd(x);  
326.         }  
327. void increment()  
328.         {  
329.             incrementAndGet();  
330.         }  
331. void decrement()  
332.         {  
333.             getAndAdd(-1);  
334.         }  
335.         T getAndSet(T newValue)  
336.         {  
337. return __sync_lock_test_and_set(&value_, newValue);  
338.         }  
339. private:  
340. volatile T value_;  
341. };  
342. typedef AtomicIntegerT<int32_t> AtomicInt32;  
343. typedef AtomicIntegerT<int64_t> AtomicInt64;  
345. class Channel;//前向声明，事件分发器主要用于事件注册与事件处理(事件回调)
346. class Poller;//IO复用机制，主要功能是监听事件集合，即select，poll,epoll的功能
347. class Timer;  
348. class TimerId;  
349. class Timestamp;  
350. class TimerQueue;  
351. class TcpConnection;  
352. class Buffer;  
354. typedef shared_ptr<TcpConnection> TcpConnectionPtr;  
355. typedef function<void()> TimerCallback;  
356. typedef function<void (const TcpConnectionPtr&)> ConnectionCallback;  
357. typedef function<void (const TcpConnectionPtr&,Buffer* buf)> MessageCallback;  
358. typedef function<void (const TcpConnectionPtr&)> WriteCompleteCallback;  
359. typedef function<void (const TcpConnectionPtr&)> CloseCallback;  
360. /*
361. *EventLoop: 事件循环，一个线程一个事件循环即one loop per thread，其主要功能是运行事件循环如等待事件发生然后处理发生的事件
362. */
363. class EventLoop:noncopyable{  
364. public:  
365. //实现事件循环
366. //实现定时回调功能，通过timerfd和TimerQueue实现
367. //实现用户任务回调，为了线程安全有可能其它线程向IO线程的EventLoop添加任务，此时通过eventfd通知EventLoop执行用户任务
368. typedef function<void()> Functor;//回调函数
369.         EventLoop();  
370.         ~EventLoop();  
371. void loop();//EventLoop的主体,用于事件循环，Eventloop::loop()->Poller::Poll()获得就绪的事件集合并通过Channel::handleEvent()执行就绪事件回调
372. void quit();//终止事件循环，通过设定标志位所以有一定延迟
373. //Timestamp pollReturnTime() const;
374. void assertInLoopThread(){//若运行线程不拥有EventLoop则退出，保证one loop per thread
375. if(!isInLoopThread()){  
376.                 abortNotInLoopThread();  
377.             }  
378.         }  
379. bool isInLoopThread() const{return threadID_==syscall(SYS_gettid);}//判断运行线程是否为拥有此EventLoop的线程
380.         TimerId runAt(const Timestamp& time,const TimerCallback& cb);//绝对时间执行定时器回调cb
381.         TimerId runAfter(double delay,const TimerCallback& cb);//相对时间执行定时器回调
382.         TimerId runEvery(double interval,const TimerCallback& cb);//每隔interval执行定时器回调
383. void runInLoop(const Functor& cb);//用于IO线程执行用户回调(如EventLoop由于执行事件回调阻塞了，此时用户希望唤醒EventLoop执行用户指定的任务)
384. void queueInLoop(const Functor& cb);//唤醒IO线程(拥有此EventLoop的线程)并将用户指定的任务回调放入队列
385. void cancel(TimerId tiemrId);  
386. void wakeup();//唤醒IO线程
387. void updateChannel(Channel* channel);//更新事件分发器Channel，完成文件描述符fd向事件集合注册事件及事件回调函数
388. void removeChannel(Channel* channel);  
389. private:  
390. void abortNotInLoopThread();//在不拥有EventLoop线程中终止
391. void handleRead();//timerfd上可读事件回调
392. void doPendingFunctors();//执行队列pendingFunctors中的用户任务回调
393. typedef vector<Channel*> ChannelList;//事件分发器Channel容器，一个Channel只负责一个文件描述符fd的事件分发
394. bool looping_;//事件循环主体loop是运行标志
395. bool quit_;//取消循环主体标志
396. const pid_t threadID_;//EventLoop的附属线程ID
397.         scoped_ptr<Poller> poller_;//IO复用器Poller用于监听事件集合
398. //scoped_ptr<Epoller> poller_;
399.         ChannelList activeChannels_;//类似与poll的就绪事件集合，这里集合换成Channel(事件分发器具备就绪事件回调功能)
400. //Timestamp pollReturnTime_;
401. int wakeupFd_;//eventfd用于唤醒EventLoop所在线程
402.         scoped_ptr<Channel> wakeupChannel_;//通过wakeupChannel_观察wakeupFd_上的可读事件，当可读时表明需要唤醒EventLoop所在线程执行用户回调
403.         Mutex mutex_;//互斥量用以保护队列
404.         vector<Functor> pendingFunctors_;//用户任务回调队列
405.         scoped_ptr<TimerQueue> timerQueue_;//定时器队列用于存放定时器
406. bool callingPendingFunctors_;//是否有用户任务回调标志
407. };  
409. /*
410.  *Poller: IO Multiplexing Poller即poll的封装，主要完成事件集合的监听
411.  */
412. class Poller:noncopyable{//生命期和EventLoop一样长，不拥有Channel
413. public:  
414. typedef vector<Channel*> ChannelList;//Channel容器(Channel包含了文件描述符fd和fd注册的事件及事件回调函数)，Channel包含文件描述符及其注册事件及其事件回调函数，这里主要用于返回就绪事件集合
415.         Poller(EventLoop* loop);  
416.         ~Poller();  
417.         Timestamp Poll(int timeoutMs,ChannelList* activeChannels);//Poller的核心功能，通过poll系统调用将就绪事件集合通过activeChannels返回，并EventLoop::loop()->Channel::handelEvent()执行相应的就绪事件回调
418. void updateChannel(Channel* channel);//Channel::update(this)->EventLoop::updateChannel(Channel*)->Poller::updateChannel(Channel*)负责维护和更新pollfs_和channels_,更新或添加Channel到Poller的pollfds_和channels_中(主要是文件描述符fd对应的Channel可能想修改已经向poll注册的事件或者fd想向poll注册事件)
419. void assertInLoopThread(){//判定是否和EventLoop的隶属关系，EventLoop要拥有此Poller
420.             ownerLoop_->assertInLoopThread();  
421.         }  
422. void removeChannel(Channel* channel);//通过EventLoop::removeChannel(Channel*)->Poller::removeChannle(Channel*)注销pollfds_和channels_中的Channel
423. private:  
424. void fillActiveChannels(int numEvents,ChannelList* activeChannels) const;//遍历pollfds_找出就绪事件的fd填入activeChannls,这里不能一边遍历pollfds_一边执行Channel::handleEvent()因为后者可能添加或者删除Poller中含Channel的pollfds_和channels_(遍历容器的同时存在容器可能被修改是危险的),所以Poller仅仅是负责IO复用，不负责事件分发(交给Channel处理)
425. typedef vector<struct pollfd> PollFdList;//struct pollfd是poll系统调用监听的事件集合参数
426. typedef map<int,Channel*> ChannelMap;//文件描述符fd到IO分发器Channel的映射，通过fd可以快速找到Channel
427. //注意:Channel中有fd成员可以完成Channel映射到fd的功能，所以fd和Channel可以完成双射
428.         EventLoop* ownerLoop_;//隶属的EventLoop
429.         PollFdList pollfds_;//监听事件集合
430.         ChannelMap channels_;//文件描述符fd到Channel的映射
431. };  
433. /*
434.  *Channel: 事件分发器,该类包含：文件描述符fd、fd欲监听的事件、事件的处理函数(事件回调函数)
435.  */
436. class Channel:noncopyable{  
437. public:  
438. typedef function<void()> EventCallback;//事件回调函数类型,回调函数的参数为空，这里将参数类型已经写死了
439. typedef function<void()> ReadEventCallback;  
440.         Channel(EventLoop* loop,int fd);//一个Channel只负责一个文件描述符fd但Channel不拥有fd，可见结构应该是这样的：EventLoop调用Poller监听事件集合，就绪的事件集合元素就是Channel。但Channel的功能不仅是返回就绪事件，还具备事件处理功能
441.         ~Channel();//目前缺失一个功能：~Channel()->EventLoop::removeChannel()->Poller::removeChannel()注销Poller::map<int,Channel*>的Channel*避免空悬指针
442. void handleEvent();//这是Channel的核心，当fd对应的事件就绪后Channel::handleEvent()执行相应的事件回调，如可读事件执行readCallback_()
443. void setReadCallback(const ReadEventCallback& cb){//可读事件回调
444.             readCallback_=cb;  
445.         }  
446. void setWriteCallback(const EventCallback& cb){//可写事件回调
447.             writeCallback_=cb;  
448.         }  
449. void setErrorCallback(const EventCallback& cb){//出错事件回调
450.             errorCallback_=cb;  
451.         }  
452. void setCloseCallback(const EventCallback& cb){  
453.             closeCallback_=cb;  
454.         }  
455. int fd() const{return fd_;}//返回Channel负责的文件描述符fd，即建立Channel到fd的映射
456. int events() const{return events_;}//返回fd域注册的事件类型
457. void set_revents(int revt){//设定fd的就绪事件类型，再poll返回就绪事件后将就绪事件类型传给此函数，然后此函数传给handleEvent，handleEvent根据就绪事件的类型决定执行哪个事件回调函数
458.             revents_=revt;  
459.         }  
460. bool isNoneEvent() const{//fd没有想要注册的事件
461. return events_==kNoneEvent;  
462.         }  
463. void enableReading(){//fd注册可读事件
464.             events_|=kReadEvent;  
465.             update();  
466.         }  
467. void enableWriting(){//fd注册可写事件
468.             events_|=kWriteEvent;  
469.             update();  
470.         }  
471. void disableWriting(){  
472.             events_&=~kWriteEvent;  
473.             update();  
474.         }  
475. void disableAll(){events_=kReadEvent;update();}  
476. bool isWriting() const{  
477. return events_&kWriteEvent;  
478.         }  
479. int index(){return index_;}//index_是本Channel负责的fd在poll监听事件集合的下标，用于快速索引到fd的pollfd
480. void set_index(int idx){index_=idx;}  
481.         EventLoop* ownerLoop(){return loop_;}  
482. private:  
483. void update();//Channel::update(this)->EventLoop::updateChannel(Channel*)->Poller::updateChannel(Channel*)最后Poller修改Channel，若Channel已经存在于Poller的vector<pollfd> pollfds_(其中Channel::index_是vector的下标)则表明Channel要重新注册事件，Poller调用Channel::events()获得事件并重置vector中的pollfd;若Channel没有在vector中则向Poller的vector添加新的文件描述符事件到事件表中，并将vector.size(),(vector每次最后追加)，给Channel::set_index()作为Channel记住自己在Poller中的位置
484. static const int kNoneEvent;//无任何事件
485. static const int kReadEvent;//可读事件
486. static const int kWriteEvent;//可写事件
487. bool eventHandling_;  
488.         EventLoop* loop_;//Channel隶属的EventLoop(原则上EventLoop，Poller，Channel都是一个IO线程)
489. const int fd_;//每个Channel唯一负责的文件描述符，Channel不拥有fd
490. int events_;//fd_注册的事件
491. int revents_;//通过poll返回的就绪事件类型
492. int index_;//在poll的监听事件集合pollfd的下标，用于快速索引到fd的pollfd
493.         ReadEventCallback readCallback_;//可读事件回调函数，当poll返回fd_的可读事件时调用此函数执行相应的事件处理，该函数由用户指定
494.         EventCallback writeCallback_;//可写事件回调函数
495.         EventCallback errorCallback_;//出错事件回调函数
496.         EventCallback closeCallback_;  
497. };  
499. /*
500. *时间戳，采用一个整数表示微秒数
501. */
502. class Timestamp{  
503. public:  
504.         Timestamp():microSecondsSinceEpoch_(0){}  
505. explicit Timestamp(int64_t microseconds):microSecondsSinceEpoch_(microseconds){}  
506. void swap(Timestamp& that){  
507.             std::swap(microSecondsSinceEpoch_,that.microSecondsSinceEpoch_);  
508.         }  
509. bool valid() const{return microSecondsSinceEpoch_>0;}  
510.         int64_t microSecondsSinceEpoch() const {return microSecondsSinceEpoch_;}  
511. static Timestamp now(){  
512. struct timeval tv;  
513.             gettimeofday(&tv, NULL);  
514.             int64_t seconds = tv.tv_sec;  
515. return Timestamp(seconds * kMicroSecondsPerSecond + tv.tv_usec);  
516.         }  
517. static Timestamp invalid(){return Timestamp();}  
518. static const int kMicroSecondsPerSecond=1000*1000;  
519. private:  
520.         int64_t microSecondsSinceEpoch_;  
521. };  
522. //时间戳的比较
523. inline bool operator<(Timestamp lhs, Timestamp rhs)  
524. {  
525. return lhs.microSecondsSinceEpoch() < rhs.microSecondsSinceEpoch();  
526. }  
528. inline bool operator==(Timestamp lhs, Timestamp rhs)  
529. {  
530. return lhs.microSecondsSinceEpoch() == rhs.microSecondsSinceEpoch();  
531. }  
532. inline double timeDifference(Timestamp high, Timestamp low)  
533. {  
534.   int64_t diff = high.microSecondsSinceEpoch() - low.microSecondsSinceEpoch();  
535. return static_cast<double>(diff) / Timestamp::kMicroSecondsPerSecond;  
536. }  
537. inline Timestamp addTime(Timestamp timestamp, double seconds)  
538. {  
539.   int64_t delta = static_cast<int64_t>(seconds * Timestamp::kMicroSecondsPerSecond);  
540. return Timestamp(timestamp.microSecondsSinceEpoch() + delta);  
541. }  
542. /*
543.  * TimerId带有唯一序号的Timer
544.  */
545. class TimerId{  
546. public:  
547.         TimerId(Timer* timer=NULL,int64_t seq=0)  
548.             :timer_(timer),sequence_(seq){}  
549. friend class TimerQueue;  
550. private:  
551.         Timer* timer_;  
552.         int64_t sequence_;  
553. };  
554. /*
555.  *定时器
556.  */
557. class Timer : boost::noncopyable  
558. {  
559. public:  
560. typedef function<void()> TimerCallback;//定时器回调函数
561. //typedef function<void()> callback;
562.         Timer(const TimerCallback& cb, Timestamp when, double interval)  
563.             :callback_(cb),expiration_(when),  
564.             interval_(interval),repeat_(interval > 0.0),  
565.             sequence_(s_numCreated_.incrementAndGet()){}  
566. void run() const {//执行定时器回调
567.             callback_();  
568.         }  
569.         Timestamp expiration() const  { return expiration_; }//返回定时器的超时时间戳
570. bool repeat() const { return repeat_; }//是否周期性定时
571.         int64_t sequence() const{return sequence_;}  
572. void restart(Timestamp now);//重置定时器
573. private:  
574. const TimerCallback callback_;//超时回调函数
575.         Timestamp expiration_;//超时时间戳
576. const double interval_;//相对时间，作为参数传给时间戳生成具体的超时时间
577. const bool repeat_;//是否重复定时标志
578. const int64_t sequence_;//
579. static AtomicInt64 s_numCreated_;//原子操作，用于生成定时器ID
580. };  
581. AtomicInt64 Timer::s_numCreated_;  
582. void Timer::restart(Timestamp now){  
583. if (repeat_){//周期定时
584.         expiration_ = addTime(now, interval_);  
585.     }  
586. else{  
587.         expiration_ = Timestamp::invalid();  
588.     }  
589. }  
590. /*
591.  * timerfd的相关操作,可用于TimerQueue实现超时器管理
592.  */
593. int createTimerfd(){//创建timerfd
594. int timerfd = ::timerfd_create(CLOCK_MONOTONIC,TFD_NONBLOCK | TFD_CLOEXEC);  
595. if (timerfd < 0){  
596.         printf("Timderfd::create() error\n");  
597.     }  
598. return timerfd;  
599. }  
600. struct timespec howMuchTimeFromNow(Timestamp when){  
601.     int64_t microseconds = when.microSecondsSinceEpoch()- Timestamp::now().microSecondsSinceEpoch();  
602. if (microseconds < 100){  
603.         microseconds = 100;  
604.     }  
605. struct timespec ts;  
606.     ts.tv_sec = static_cast<time_t>(microseconds / Timestamp::kMicroSecondsPerSecond);  
607.     ts.tv_nsec = static_cast<long>((microseconds % Timestamp::kMicroSecondsPerSecond) * 1000);  
608. return ts;  
609. }  
610. void readTimerfd(int timerfd, Timestamp now){//timerfd的可读事件回调
611.     uint64_t howmany;  
612.     ssize_t n = ::read(timerfd, &howmany, sizeof howmany);  
613. if (n != sizeof howmany){  
614.         printf("readTimerfd error\n");  
615.     }  
616. }  
617. void resetTimerfd(int timerfd, Timestamp expiration)//重置timerfd的计时
618. {  
619. struct itimerspec newValue;  
620. struct itimerspec oldValue;  
621.     bzero(&newValue, sizeof newValue);  
622.     bzero(&oldValue, sizeof oldValue);  
623.     newValue.it_value = howMuchTimeFromNow(expiration);  
624. int ret = ::timerfd_settime(timerfd, 0, &newValue, &oldValue);  
625. if (ret){  
626.         printf("timerfd_settime erro\n");  
627.     }  
628. }  
629. /*
630.  *定时器队列
631.  */
632. class TimerQueue : boost::noncopyable  
633. {//其通过添加一个timerfd到EventLoop中，当timerfd可读事件就绪时，TimerQueue::handleRead()遍历容器内的超时的定时器并执行这些超时定时器的回调
634. //定时器容器为set<pair<Timestamp,Timer*> >，采用pair作为key的原因是可能在一个时刻有多个相同的Timestamp时间戳
635. //当向定时器容器set添加定时器timer的时候会检查当前最小的定时器，并将最小的定时时间付赋给timerfd
636. public:  
637. typedef function<void()> TimerCallback;//定时器回调
638.         TimerQueue(EventLoop* loop);  
639.         ~TimerQueue();  
640.         TimerId addTimer(const TimerCallback& cb,Timestamp when,double interval);//添加定时器到定时器队列中
641. void cancel(TimerId timerId);  
642. private:  
643. typedef pair<Timestamp, Timer*> Entry;//采用此作为键值
644. typedef set<Entry> TimerList;//set只有key无value且有序
645. typedef pair<Timer*,int64_t> ActiveTimer;  
646. typedef set<ActiveTimer> ActiveTimerSet;  
648. void handleRead();//timerfd的可读回调
649. void addTimerInLoop(Timer* timer);//添加定时器
650. void cancelInLoop(TimerId timerId);  
651.         std::vector<Entry> getExpired(Timestamp now);//获取所有超时的定时器
652. void reset(const std::vector<Entry>& expired, Timestamp now);//超时的定时器是否需要重新定时
653. bool insert(Timer* timer);//插入定时器到队列中
655.         EventLoop* loop_;//TimerQueue所属的EventLoop
656. const int timerfd_;//定时器队列本身需要在定时器超时后执行队列中所有超时定时器的回调
657.         Channel timerfdChannel_;//采用timerfdChannel_观察timerfd_的可读事件啊，当timerfd_可读表明定时器队列中有定时器超时
658.         TimerList timers_;//定时器队列
659. bool callingExpiredTimers_;  
660.         ActiveTimerSet activeTimers_;  
661.         ActiveTimerSet cancelingTimers_;  
662. };  
663. /*
664.  * TimerQueue实现
665.  */
666. TimerQueue::TimerQueue(EventLoop* loop)  
667.     :loop_(loop),timerfd_(createTimerfd()),  
668.     timerfdChannel_(loop, timerfd_),timers_(),  
669.     callingExpiredTimers_(false)  
670. {  
671.     timerfdChannel_.setReadCallback(bind(&TimerQueue::handleRead, this));  
672.     timerfdChannel_.enableReading();//timerfd注册可读事件
673. }  
674. TimerQueue::~TimerQueue(){  
675.     ::close(timerfd_);  
676. for (TimerList::iterator it = timers_.begin();it != timers_.end(); ++it)  
677.     {  
678. delete it->second;  
679.     }  
680. }  
681. TimerId TimerQueue::addTimer(const TimerCallback& cb,Timestamp when,double interval)//其它线程向IO线程添加用户回调时将添加操作转移到IO线程中去，从而保证线程安全one loop per thread
682. {//由EventLoop::runAt等函数调用
683.     Timer* timer = new Timer(cb, when, interval);  
684.     loop_->runInLoop(bind(&TimerQueue::addTimerInLoop,this,timer));//通过EventLoop::runInLoop()->TimerQueue::queueInLoop()
685. //runInLoop语义是若是本IO线程想要添加定时器则直接由addTimerInLoop添加，若是其它线程向IO线程添加定时器则需要间接通过queueInLoop添加
686. return TimerId(timer,timer->sequence());  
687. }  
688. void TimerQueue::addTimerInLoop(Timer* timer){//IO线程自己向自己添加定时器
689.     loop_->assertInLoopThread();  
690. bool earliestChanged=insert(timer);//若当前插入的定时器比队列中的定时器都早则返回真
691. if(earliestChanged){  
692.         resetTimerfd(timerfd_,timer->expiration());//timerfd重新设置超时时间
693.     }  
694. }  
695. void TimerQueue::cancel(TimerId timerId){  
696.     loop_->runInLoop(bind(&TimerQueue::cancelInLoop,this,timerId));  
697. }  
698. void TimerQueue::cancelInLoop(TimerId timerId){  
699.     loop_->assertInLoopThread();  
700.     assert(timers_.size()==activeTimers_.size());  
701.     ActiveTimer timer(timerId.timer_,timerId.sequence_);  
702.     ActiveTimerSet::iterator it=activeTimers_.find(timer);  
703. if(it!=activeTimers_.end()){  
704. size_t n=timers_.erase(Entry(it->first->expiration(),it->first));  
705.         assert(n==1);  
706.         (void)n;  
707. delete it->first;  
708.         activeTimers_.erase(it);  
709.     }  
710. else if(callingExpiredTimers_){  
711.         cancelingTimers_.insert(timer);  
712.     }  
713.     assert(timers_.size()==activeTimers_.size());  
714. }  
715. void TimerQueue::handleRead(){//timerfd的回调函数
716.     loop_->assertInLoopThread();  
717.     Timestamp now(Timestamp::now());  
718.     readTimerfd(timerfd_, now);  
719.     std::vector<Entry> expired = getExpired(now);//TimerQueue::timerfd可读表明队列中有定时器超时，则需要找出那些超时的定时器
720.     callingExpiredTimers_=true;  
721.     cancelingTimers_.clear();  
722. for (std::vector<Entry>::iterator it = expired.begin();it!= expired.end(); ++it)//
723.     {  
724.         it->second->run();//执行定时器Timer的超时回调
725.     }  
726.     callingExpiredTimers_=false;  
727.     reset(expired, now);//查看已经执行完的超市定时器是否需要再次定时
728. }  
729. std::vector<TimerQueue::Entry> TimerQueue::getExpired(Timestamp now)//获取队列中的超时的定时器(可能多个)
730. {  
731.     assert(timers_.size()==activeTimers_.size());  
732.     std::vector<Entry> expired;  
733.     Entry sentry = std::make_pair(now, reinterpret_cast<Timer*>(UINTPTR_MAX));  
734.     TimerList::iterator it = timers_.lower_bound(sentry);//返回比参数小的下界，即返回第一个当前未超时的定时器(可能没有这样的定时器)
735. //lower_bound(value_type& val)调用key_comp返回第一个不小于val的迭代器
736.     assert(it == timers_.end() || now < it->first);  
737.     std::copy(timers_.begin(), it, back_inserter(expired));  
738.     timers_.erase(timers_.begin(), it);  
739.     BOOST_FOREACH(Entry entry,expired){  
740.         ActiveTimer timer(entry.second,entry.second->sequence());  
741. size_t n=activeTimers_.erase(timer);  
742.         assert(n==1);  
743.         (void)n;  
744.     }  
745. return expired;//返回已经超时的那部分定时器
746. }  
748. void TimerQueue::reset(const std::vector<Entry>& expired, Timestamp now)//已经执行完超时回调的定时器是否需要重置定时
749. {  
750.     Timestamp nextExpire;  
751. for (std::vector<Entry>::const_iterator it = expired.begin();it != expired.end(); ++it)  
752.     {  
753.         ActiveTimer timer(it->second,it->second->sequence());  
754. if (it->second->repeat()&&  
755.                 cancelingTimers_.find(timer)==cancelingTimers_.end()){//需要再次定时
756.             it->second->restart(now);  
757.             insert(it->second);  
758.         }  
759. else{//否则删除该定时器
760. delete it->second;  
761.         }  
762.     }  
763. if (!timers_.empty()){//为超时定时器重新定时后需要获取当前最小的超时时间给timerfd，以防重置的这些超市定时器中含有最小的超时时间
764.         nextExpire = timers_.begin()->second->expiration();  
765.     }  
766. if (nextExpire.valid()){  
767.         resetTimerfd(timerfd_, nextExpire);//重置timerfd的超时时间
768.     }  
769. }  
770. bool TimerQueue::insert(Timer* timer)//向超时队列中插入定时器
771. {  
772.     loop_->assertInLoopThread();  
773.     assert(timers_.size()==activeTimers_.size());  
774. bool earliestChanged = false;  
775.     Timestamp when = timer->expiration();  
776.     TimerList::iterator it = timers_.begin();  
777. if (it == timers_.end() || when < it->first)  
778.     {  
779.         earliestChanged = true;//当前插入的定时器是队列中最小的定时器，此时外层函数需要重置timerfd的超时时间
780.     }  
781.     {  
782.         pair<TimerList::iterator,bool> result=  
783.             timers_.insert(Entry(when,timer));  
784.         assert(result.second);  
785.         (void)result;  
786.     }  
787.     {  
788.         pair<ActiveTimerSet::iterator,bool> result=  
789.             activeTimers_.insert(ActiveTimer(timer,timer->sequence()));  
790.         assert(result.second);  
791.         (void)result;  
792.     }  
793.     assert(timers_.size()==activeTimers_.size());  
794. return earliestChanged;  
795. }  
797. /*
798. *EventLoop成员实现
799. */
800. class IngnoreSigPipe{  
801. public:  
802.         IngnoreSigPipe(){  
803.             ::signal(SIGPIPE,SIG_IGN);  
804.         }  
805. };  
806. IngnoreSigPipe initObj;  
807. __thread EventLoop* t_loopInThisThread=0;//线程私有数据表示线程是否拥有EventLoop
808. const int kPollTimeMs=10000;//poll等待时间
809. static int createEventfd(){//创建eventfd，eventfd用于唤醒
810. int evtfd=eventfd(0,EFD_NONBLOCK|EFD_CLOEXEC);  
811. if(evtfd<0){  
812.         printf("Failed in eventfd\n");  
813.         abort();  
814.     }  
815. return evtfd;  
816. }  
817. EventLoop::EventLoop()  
818.     :looping_(false),  
819.     quit_(false),  
820.     threadID_(syscall(SYS_gettid)),  
821.     poller_(new Poller(this)),  
822.     timerQueue_(new TimerQueue(this)),//EventLoop用于一个定时器队列
823.     wakeupFd_(createEventfd()),  
824.     wakeupChannel_(new Channel(this,wakeupFd_)),//通过Channel观察wakeupFd_
825.     callingPendingFunctors_(false)  
826. {  
827. if(!t_loopInThisThread){  
828.         t_loopInThisThread=this;//EventLoop构造时线程私有数据记录
829.     }  
830.     wakeupChannel_->setReadCallback(bind(&EventLoop::handleRead,this));//设置eventfd的回调
831.     wakeupChannel_->enableReading();//eventfd的可读事件,并Channel::update(this)将eventfd添加到poll事件表中
832. }  
833. EventLoop::~EventLoop(){  
834.     assert(!looping_);  
835.     close(wakeupFd_);  
836.     t_loopInThisThread=NULL;//EventLoop析构将其置空
837. }  
838. void EventLoop::loop(){//EventLoop主循环，主要功能是监听事件集合，执行就绪事件的处理函数
839.     assert(!looping_);  
840.     assertInLoopThread();  
841.     looping_=true;  
842.     quit_=false;  
843. while(!quit_){  
844.         activeChannels_.clear();  
845.         poller_->Poll(kPollTimeMs,&activeChannels_);//activeChannels是就绪事件
846. for(ChannelList::iterator it=activeChannels_.begin();it!=activeChannels_.end();it++){  
847.             (*it)->handleEvent();//处理就绪事件的回调函数，处理事件回调
848.         }  
849.         doPendingFunctors();//处理用户任务回调
850.     }  
851.     looping_=false;  
852. }  
853. void EventLoop::quit(){  
854.     quit_=true;//停止主循环标志，主循环不会马上停止有延迟
855. if(!isInLoopThread()){  
856.         wakeup();//其它线程唤醒EventLoop线程且终止它
857.     }  
858. }  
859. void EventLoop::updateChannel(Channel* channel){//主要用于文件描述符添加到poll的监听事件集合中
860.     assert(channel->ownerLoop()==this);  
861.     assertInLoopThread();  
862.     poller_->updateChannel(channel);  
863. }  
864. void EventLoop::abortNotInLoopThread(){  
865.     printf("abort not in Loop Thread\n");  
866.     abort();//非本线程调用强行终止
867. }  
868. TimerId EventLoop::runAt(const Timestamp& time, const TimerCallback& cb)//绝对时间执行回调
869. {  
870. return timerQueue_->addTimer(cb, time, 0.0);  
871. }  
872. TimerId EventLoop::runAfter(double delay, const TimerCallback& cb)//相对时间执行回调
873. {  
874.     Timestamp time(addTime(Timestamp::now(), delay));  
875. return runAt(time, cb);  
876. }  
877. TimerId EventLoop::runEvery(double interval, const TimerCallback& cb)//周期性回调
878. {  
879.     Timestamp time(addTime(Timestamp::now(), interval));//Timestamp::addTime
880. return timerQueue_->addTimer(cb, time, interval);  
881. }  
882. void EventLoop::cancel(TimerId timerId){  
883. return timerQueue_->cancel(timerId);  
884. }  
885. void EventLoop::runInLoop(const Functor& cb){  
886. if(isInLoopThread()){//本IO线程调用则直接执行执行用户回调
887.        cb();  
888.     }  
889. else{//其它线程调用runInLoop则向用户回调队列添加，保证线程安全one loop per thread
890.         queueInLoop(cb);  
891.     }  
892. }  
893. void EventLoop::queueInLoop(const Functor& cb){  
894.     {  
895.         MutexLockGuard lock(mutex_);//互斥量保护用户回调队列
896.         pendingFunctors_.push_back(cb);  
897.     }  
898. if(!isInLoopThread()||callingPendingFunctors_){  
899.         wakeup();//其它线程添加用户回调任务或者EventLoop的IO线程正在处理用户任务回调时，若阻塞则唤醒IO线程
900.     }  
901. }  
902. void EventLoop::handleRead(){//eventfd可读回调
903.     uint64_t one=1;  
904.     ssize_t n=read(wakeupFd_,&one,sizeof(one));  
905. if(n!=sizeof(one)){  
906.         printf("EventLoop::handleRead() error\n");  
907.     }  
908. }  
909. void EventLoop::doPendingFunctors(){//执行用户任务回调
910.     vector<Functor> functors;  
911.     callingPendingFunctors_=true;  
912.     {  
913.         MutexLockGuard lock(mutex_);  
914.         functors.swap(pendingFunctors_);//采用swap而不是在这里执行回调是为了缩小临界区
915.     }  
916. for(size_t i=0;i<functors.size();i++){  
917.         functors[i]();  
918.     }  
919.     callingPendingFunctors_=false;  
920. }  
921. void EventLoop::wakeup(){  
922.     uint64_t one=1;  
923.     ssize_t n=write(wakeupFd_,&one,sizeof(one));//通过eventfd通知
924. if(n!=sizeof(one)){  
925.         printf("EventLoop::wakeup() write error\n");  
926.     }  
927. }  
928. void EventLoop::removeChannel(Channel* channel){  
929.     assert(channel->ownerLoop()==this);  
930.     assertInLoopThread();  
931.     poller_->removeChannel(channel);  
932. }  
934. /*
935. *Poller成员实现
936. */
937. Poller::Poller(EventLoop* loop):ownerLoop_(loop){}//Poller明确所属的EventLoop
938. Poller::~Poller(){}  
939. Timestamp Poller::Poll(int timeoutMs,ChannelList* activeChannels){  
940. int numEvents=poll(&*pollfds_.begin(),pollfds_.size(),timeoutMs);//poll监听事件集合pollfds_
941.     Timestamp now(Timestamp::now());  
942. if(numEvents>0){  
943.         fillActiveChannels(numEvents,activeChannels);//将就绪的事件添加到activeChannels
944.     }  
945. else if(numEvents==0){  
946.     }  
947. else{  
948.         printf("Poller::Poll error\n");  
949.     }  
950. return now;  
951. }  
952. void Poller::fillActiveChannels(int numEvents,ChannelList* activeChannels) const{//将就绪事件通过activeChannels返回
953. for(PollFdList::const_iterator pfd=pollfds_.begin();pfd!=pollfds_.end()&&numEvents>0;++pfd){  
954. if(pfd->revents>0){  
955.             --numEvents;//若numEvents个事件全部找到就不需要再遍历容器剩下的部分
956.             ChannelMap::const_iterator ch=channels_.find(pfd->fd);  
957.             assert(ch!=channels_.end());  
958.             Channel* channel=ch->second;  
959.             assert(channel->fd()==pfd->fd);  
960.             channel->set_revents(pfd->revents);  
961.             activeChannels->push_back(channel);  
962.         }  
963.     }  
964. }  
965. void Poller::updateChannel(Channel* channel){  
966.     assertInLoopThread();  
967. if(channel->index()<0){//若channel的文件描述符fd没有添加到poll的监听事件集合中
968.         assert(channels_.find(channel->fd())==channels_.end());  
969. struct pollfd pfd;  
970.         pfd.fd=channel->fd();  
971.         pfd.events=static_cast<short>(channel->events());  
972.         pfd.revents=0;  
973.         pollfds_.push_back(pfd);  
974. int idx=static_cast<int>(pollfds_.size())-1;  
975.         channel->set_index(idx);  
976.         channels_[pfd.fd]=channel;  
977.     }  
978. else{//若已经添加到监听事件集合中，但是需要修改
979.         assert(channels_.find(channel->fd())!=channels_.end());  
980.         assert(channels_[channel->fd()]==channel);  
981. int idx=channel->index();  
982.         assert(0<=idx&&idx<static_cast<int>(pollfds_.size()));  
983. struct pollfd& pfd=pollfds_[idx];  
984.         assert(pfd.fd==channel->fd()||pfd.fd==-channel->fd()-1);//pfd.fd=-channel->fd()-1是为了让poll忽略那些kNoneEvent的描述符，-1是因为:fd可能为0所以-channel->fd()可能还是0,不能区分一个不可能的描述符
985.         pfd.events=static_cast<short>(channel->events());//修改注册事件类型
986.         pfd.revents=0;  
987. if(channel->isNoneEvent()){  
988.             pfd.fd=-channel->fd()-1;//channel::events_=kNoneEvent时poll忽略那些不可能的描述符-channel->fd()-1,-1原因见上面
989.         }  
990.     }  
991. }  
992. void Poller::removeChannel(Channel* channel)  
993. {  
994.   assertInLoopThread();  
995.   assert(channels_.find(channel->fd()) != channels_.end());  
996.   assert(channels_[channel->fd()] == channel);  
997.   assert(channel->isNoneEvent());  
998. int idx = channel->index();  
999.   assert(0 <= idx && idx < static_cast<int>(pollfds_.size()));  
1000. const struct pollfd& pfd = pollfds_[idx]; (void)pfd;  
1001.   assert(pfd.fd == -channel->fd()-1 && pfd.events == channel->events());  
1002. size_t n = channels_.erase(channel->fd());  
1003.   assert(n == 1); (void)n;  
1004. if (implicit_cast<size_t>(idx) == pollfds_.size()-1) {  
1005.     pollfds_.pop_back();  
1006.   } else {  
1007. int channelAtEnd = pollfds_.back().fd;  
1008.     iter_swap(pollfds_.begin()+idx, pollfds_.end()-1);  
1009. if (channelAtEnd < 0) {  
1010.       channelAtEnd = -channelAtEnd-1;  
1011.     }  
1012.     channels_[channelAtEnd]->set_index(idx);  
1013.     pollfds_.pop_back();  
1014.   }  
1015. }  
1016. /*
1017. *Channel成员实现
1018. */
1019. const int Channel::kNoneEvent=0;//无事件
1020. const int Channel::kReadEvent=POLLIN|POLLPRI;//可读事件
1021. const int Channel::kWriteEvent=POLLOUT;//可写事件
1022. Channel::Channel(EventLoop* loop,int fdArg)  
1023.     :loop_(loop),fd_(fdArg),events_(0),revents_(0),  
1024.     index_(-1),eventHandling_(false)  
1025.     {}  
1026. void Channel::update(){//添加或修改文件描述符的事件类型
1027.     loop_->updateChannel(this);  
1028. }  
1029. Channel::~Channel(){  
1030.     assert(!eventHandling_);  
1031. }  
1032. void Channel::handleEvent(){//处理就绪事件的处理函数
1033.     eventHandling_=true;  
1034. if(revents_&POLLNVAL){  
1035.         printf("Channel::handleEvent() POLLNVAL\n");  
1036.     }  
1037. if((revents_&POLLHUP)&&!(revents_&POLLIN)){//出错回调
1038.         printf("Channel::handle_event() POLLUP\n");  
1039. if(closeCallback_)  
1040.             closeCallback_();  
1041.     }  
1042. if(revents_&(POLLERR|POLLNVAL)){//可读回调
1043. if(errorCallback_)  
1044.             errorCallback_();  
1045.     }  
1046. if(revents_&(POLLIN|POLLPRI|POLLRDHUP)){  
1047. if(readCallback_) readCallback_();  
1048.     }  
1049. if(revents_&POLLOUT){//可写回调
1050. if(writeCallback_)  
1051.             writeCallback_();  
1052.     }  
1053.     eventHandling_=false;  
1054. }  
1056. /*
1057. *开启一个线程执行一个EventLoop，这才是one loop per thread
1058. */
1059. class EventLoopThread:noncopyable{  
1060. public:  
1061.         EventLoopThread()  
1062.             :loop_(NULL),exiting_(false),  
1063.             thread_(bind(&EventLoopThread::threadFunc,this)),  
1064.             mutex_(),cond_(mutex_){}  
1065.         ~EventLoopThread(){  
1066.             exiting_=true;  
1067.             loop_->quit();  
1068.             thread_.join();  
1069.         }  
1070.         EventLoop* startLoop(){  
1071. //assert(!thread_.started());
1072.             thread_.start();  
1073.             {  
1074.                 MutexLockGuard lock(mutex_);  
1075. while(loop_==NULL){  
1076.                     cond_.wait();  
1077.                 }  
1078.             }  
1079. return loop_;  
1080.         }  
1081. private:  
1082. void threadFunc(){  
1083.             EventLoop loop;  
1084.             {  
1085.                 MutexLockGuard lock(mutex_);  
1086.                 loop_=&loop;  
1087.                 cond_.notify();  
1088.             }  
1089.             loop.loop();  
1090.         }  
1091.         EventLoop* loop_;  
1092. bool exiting_;  
1093.         Thread thread_;  
1094.         Mutex mutex_;  
1095.         Condition cond_;  
1096. };  
1097. /*
1098.  * EventLoopThreadPool
1099.  */
1100. class EventLoopThreadPool:noncopyable{  
1101. public:  
1102.         EventLoopThreadPool(EventLoop* baseLoop)  
1103.             :baseLoop_(baseLoop),  
1104.             started_(false),numThreads_(0),next_(0){}  
1105.         ~EventLoopThreadPool(){}  
1106. void setThreadNum(int numThreads){numThreads_=numThreads;}  
1107. void start(){  
1108.             assert(!started_);  
1109.             baseLoop_->assertInLoopThread();  
1110.             started_=true;  
1111. for(int i=0;i<numThreads_;i++){  
1112.                 EventLoopThread* t=new EventLoopThread;  
1113.                 threads_.push_back(t);  
1114.                 loops_.push_back(t->startLoop());  
1115.             }  
1116.         }  
1117.         EventLoop* getNextLoop(){  
1118.             baseLoop_->assertInLoopThread();  
1119.             EventLoop* loop=baseLoop_;  
1120. if(!loops_.empty()){  
1121.                 loop=loops_[next_];  
1122.                 ++next_;  
1123. if(static_cast<size_t>(next_)>=loops_.size())  
1124.                     next_=0;  
1125.             }  
1126. return loop;  
1127.         }  
1128. private:  
1129.         EventLoop* baseLoop_;  
1130. bool started_;  
1131. int numThreads_;  
1132. int next_;  
1133.         ptr_vector<EventLoopThread> threads_;  
1134.         vector<EventLoop*> loops_;  
1135. };  
1136. /*
1137.  *常用的socket选项
1138.  */
1139. namespace sockets{  
1141. inline uint64_t hostToNetwork64(uint64_t host64)  
1142. {//主机字节序转为网络字节序
1143. return htobe64(host64);  
1144. }  
1145. inline uint32_t hostToNetwork32(uint32_t host32)  
1146. {  
1147. return htonl(host32);  
1148. }  
1149. inline uint16_t hostToNetwork16(uint16_t host16)  
1150. {  
1151. return htons(host16);  
1152. }  
1153. inline uint64_t networkToHost64(uint64_t net64)  
1154. {//网络字节序转为主机字节序
1155. return be64toh(net64);  
1156. }  
1158. inline uint32_t networkToHost32(uint32_t net32)  
1159. {  
1160. return ntohl(net32);  
1161. }  
1162. inline uint16_t networkToHost16(uint16_t net16)  
1163. {  
1164. return ntohs(net16);  
1165. }  
1167. typedef struct sockaddr SA;  
1168. const SA* sockaddr_cast(const struct sockaddr_in* addr){//强制转换
1169. return static_cast<const SA*>(implicit_cast<const void*>(addr));  
1170. }  
1171. SA* sockaddr_cast(struct sockaddr_in* addr){  
1172. return static_cast<SA*>(implicit_cast<void*>(addr));  
1173. }  
1174. void setNonBlockAndCloseOnExec(int sockfd){//将描述符设置为非阻塞和O_CLOEXEC(close on exec)
1175. int flags = ::fcntl(sockfd, F_GETFL, 0);  
1176.     flags |= O_NONBLOCK;  
1177. int ret = ::fcntl(sockfd, F_SETFL, flags);  
1178.     flags = ::fcntl(sockfd, F_GETFD, 0);  
1179.     flags |= FD_CLOEXEC;  
1180.     ret = ::fcntl(sockfd, F_SETFD, flags);  
1181. }  
1182. int createNonblockingOrDie()  
1183. {//socket()创建非阻塞的socket描述符
1184.     #if VALGRIND
1185. int sockfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);  
1186. if (sockfd < 0) {  
1187.         printf("socket() error\n");  
1188.     }  
1189.     setNonBlockAndCloseOnExec(sockfd);  
1190.     #else
1191. int sockfd = ::socket(AF_INET,  
1192.                         SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC,  
1193.                         IPPROTO_TCP);  
1194. if (sockfd < 0){  
1195.         printf("socke() error\n");  
1196.     }  
1197.     #endif
1198. return sockfd;  
1199. }  
1200. int connect(int sockfd,const struct sockaddr_in& addr){  
1201. return ::connect(sockfd,sockaddr_cast(&addr),sizeof addr);  
1202. }  
1203. void bindOrDie(int sockfd, const struct sockaddr_in& addr)  
1204. {//bind()
1205. int ret = ::bind(sockfd, sockaddr_cast(&addr), sizeof addr);  
1206. if (ret < 0) {  
1207.          printf("bind() error\n");  
1208.     }  
1209. }  
1210. void listenOrDie(int sockfd){//listen()
1211. int ret = ::listen(sockfd, SOMAXCONN);  
1212. if (ret < 0){  
1213.           printf("listen() error\n");  
1214.     }  
1215. }  
1216. int accept(int sockfd, struct sockaddr_in* addr)  
1217. {//accept()
1218.     socklen_t addrlen = sizeof *addr;  
1219.     #if VALGRIND
1220. int connfd = ::accept(sockfd, sockaddr_cast(addr), &addrlen);  
1221.     setNonBlockAndCloseOnExec(connfd);  
1222.     #else
1223. int connfd = ::accept4(sockfd, sockaddr_cast(addr),  
1224.                          &addrlen, SOCK_NONBLOCK | SOCK_CLOEXEC);  
1225.     #endif
1226. if (connfd < 0){  
1227. int savedErrno = errno;  
1228.         printf("accept error\n");  
1229. switch (savedErrno)  
1230.         {  
1231. case EAGAIN:  
1232. case ECONNABORTED:  
1233. case EINTR:  
1234. case EPROTO: // ???
1235. case EPERM:  
1236. case EMFILE: // per-process lmit of open file desctiptor ???
1237.                 errno = savedErrno;  
1238. break;  
1239. case EBADF:  
1240. case EFAULT:  
1241. case EINVAL:  
1242. case ENFILE:  
1243. case ENOBUFS:  
1244. case ENOMEM:  
1245. case ENOTSOCK:  
1246. case EOPNOTSUPP:  
1247.                 printf("accept() fatal erro\n");  
1248. break;  
1249. default:  
1250.                 printf("accpet() unknown error\n");  
1251. break;  
1252.         }  
1253.     }  
1254. return connfd;  
1255. }  
1256. void close(int sockfd){//close()
1257. if (::close(sockfd) < 0){  
1258.         printf("sockets::close\n");  
1259.     }  
1260. }  
1261. void shutdownWrite(int sockfd){  
1262. if(::shutdown(sockfd,SHUT_WR)<0)  
1263.         printf("sockets::shutdownWrite() error\n");  
1264. }  
1265. void toHostPort(char* buf, size_t size,const struct sockaddr_in& addr)  
1266. {//将IPv4地址转为IP和端口
1267. char host[INET_ADDRSTRLEN] = "INVALID";  
1268.     ::inet_ntop(AF_INET, &addr.sin_addr, host, sizeof host);  
1269.     uint16_t port =networkToHost16(addr.sin_port);  
1270.     snprintf(buf, size, "%s:%u", host, port);  
1271. }  
1272. void fromHostPort(const char* ip, uint16_t port,struct sockaddr_in* addr)  
1273. {//将主机IP和端口转为IPv4地址
1274.     addr->sin_family = AF_INET;  
1275.     addr->sin_port = hostToNetwork16(port);  
1276. if (::inet_pton(AF_INET, ip, &addr->sin_addr) <= 0)  
1277.     {  
1278.         printf("sockets::fromHostPort\n");  
1279.     }  
1280. }  
1281. sockaddr_in getLocalAddr(int sockfd)  
1282. {  
1283. struct sockaddr_in localaddr;  
1284.   bzero(&localaddr, sizeof localaddr);  
1285.   socklen_t addrlen = sizeof(localaddr);  
1286. if (::getsockname(sockfd, sockaddr_cast(&localaddr), &addrlen) < 0)  
1287.   {  
1288.       printf("getsockname() error\n");  
1289.   }  
1290. return localaddr;  
1291. }  
1292. struct sockaddr_in getPeerAddr(int sockfd){  
1293. struct sockaddr_in peeraddr;  
1294.     bzero(&peeraddr,sizeof peeraddr);  
1295.     socklen_t addrlen=sizeof peeraddr;  
1296. if(::getpeername(sockfd,sockaddr_cast(&peeraddr),&addrlen)<0)  
1297.         printf("sockets::getPeerAddr() error\n");  
1298. return peeraddr;  
1299. }  
1300. int getSocketError(int sockfd){  
1301. int optval;  
1302.     socklen_t optlen=sizeof optval;  
1303. if(getsockopt(sockfd,SOL_SOCKET,SO_ERROR,&optval,&optlen)<0){  
1304. return errno;  
1305.     }  
1306. else{  
1307. return optval;  
1308.     }  
1309. }  
1310. bool isSelfConnect(int sockfd){//自连接判断
1311. struct sockaddr_in localaddr=getLocalAddr(sockfd);  
1312. struct sockaddr_in peeraddr=getPeerAddr(sockfd);  
1313. return localaddr.sin_port==peeraddr.sin_port&&  
1314.         localaddr.sin_addr.s_addr==peeraddr.sin_addr.s_addr;  
1315. }  
1316. }//end-namespace
1317. /*
1318.  * Socket
1319.  */
1320. class InetAddress;  
1321. class Socket:noncopyable{//创建一个socket描述符fd并绑定sockaddr，监听fd功能
1322. public:  
1323. explicit Socket(uint16_t sockfd):sockfd_(sockfd){}  
1324.         ~Socket();  
1325. int fd() const{return sockfd_;}  
1326. void bindAddress(const InetAddress& addr);  
1327. void listen();  
1328. int accept(InetAddress* peeraddr);  
1329. void setReuseAddr(bool on);  
1330. void shutdownWrite(){  
1331.             sockets::shutdownWrite(sockfd_);  
1332.         }  
1333. void setTcpNoDelay(bool on){  
1334. int optval=on?1:0;  
1335.             ::setsockopt(sockfd_,IPPROTO_TCP,TCP_NODELAY,&optval,sizeof optval);  
1336.         }  
1337. private:  
1338. const int sockfd_;  
1339. };  
1340. /*
1341.  * sockaddr_in
1342.  */
1343. class InetAddress{//sockaddr地址的封装
1344. public:  
1345. explicit InetAddress(uint16_t port);  
1346.         InetAddress(const string& ip,uint16_t port);  
1347.         InetAddress(const struct sockaddr_in& addr):addr_(addr){}  
1348.         string toHostPort() const;  
1349. const struct sockaddr_in& getSockAddrInet() const{return addr_;}  
1350. void setSockAddrInet(const struct sockaddr_in& addr){addr_=addr;}  
1351. private:  
1352. struct sockaddr_in addr_;  
1353. };  
1354. BOOST_STATIC_ASSERT(sizeof(InetAddress)==sizeof(struct sockaddr_in));//编译时断言
1355. class Acceptor:noncopyable{//接受TCP连接并执行相应的回调
1356. public://Acceptor对应的是一个服务端的监听socket描述符listenfd
1357. typedef function<void(int sockfd,const InetAddress&)> NewConnectionCallback;  
1358.         Acceptor(EventLoop* loop,const InetAddress& listenAddr);  
1359. void setNewConnectionCallback(const NewConnectionCallback& cb)  
1360.         { newConnectionCallback_=cb;}  
1361. bool listening() const{return listening_;}  
1362. void listen();  
1363. private:  
1364. void handleRead();  
1365.         EventLoop* loop_;  
1366.         Socket acceptSocket_;//服务端listenfd对应RAII封装的socket描述符
1367.         Channel acceptChannel_;//采用Channel管理服务端监听端口listenfd,可以理解为Channel管理accpetSocket_里的fd
1368.         NewConnectionCallback newConnectionCallback_;  
1369. bool listening_;  
1371. };  
1372. /*
1373.  *Socket实现
1374.  */
1375. Socket::~Socket()  
1376. {  
1377.     sockets::close(sockfd_);  
1378. }  
1379. void Socket::bindAddress(const InetAddress& addr)  
1380. {  
1381.     sockets::bindOrDie(sockfd_, addr.getSockAddrInet());  
1382. }  
1383. void Socket::listen()  
1384. {  
1385.     sockets::listenOrDie(sockfd_);  
1386. }  
1387. int Socket::accept(InetAddress* peeraddr)  
1388. {  
1389. struct sockaddr_in addr;  
1390.     bzero(&addr, sizeof addr);  
1391. int connfd = sockets::accept(sockfd_, &addr);  
1392. if (connfd >= 0)  
1393.     {  
1394.         peeraddr->setSockAddrInet(addr);  
1395.     }  
1396. return connfd;  
1397. }  
1398. void Socket::setReuseAddr(bool on)  
1399. {  
1400. int optval = on ? 1 : 0;  
1401.     ::setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR,  
1402.                &optval, sizeof optval);  
1403. }  
1404. /*
1405.  *InetAddress实现
1406.  */
1407. static const in_addr_t kInaddrAny=INADDR_ANY;//任意的网络字节序IP地址为0
1408. InetAddress::InetAddress(uint16_t port)  
1409. {  
1410.     bzero(&addr_, sizeof addr_);  
1411.     addr_.sin_family = AF_INET;  
1412.     addr_.sin_addr.s_addr = sockets::hostToNetwork32(kInaddrAny);  
1413.     addr_.sin_port = sockets::hostToNetwork16(port);  
1414. }  
1415. InetAddress::InetAddress(const std::string& ip, uint16_t port)  
1416. {  
1417.     bzero(&addr_, sizeof addr_);  
1418.     sockets::fromHostPort(ip.c_str(), port, &addr_);  
1419. }  
1420. string InetAddress::toHostPort() const
1421. {  
1422. char buf[32];  
1423.     sockets::toHostPort(buf, sizeof buf, addr_);  
1424. return buf;  
1425. }  
1426. /*
1427.  *Acceptor实现
1428.  */
1429. Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr)  
1430.   : loop_(loop),  
1431.     acceptSocket_(sockets::createNonblockingOrDie()),  
1432.     acceptChannel_(loop, acceptSocket_.fd()),  
1433.     listening_(false)  
1434. {  
1435.     acceptSocket_.setReuseAddr(true);  
1436.     acceptSocket_.bindAddress(listenAddr);  
1437.     acceptChannel_.setReadCallback(  
1438.       boost::bind(&Acceptor::handleRead, this));  
1439. }  
1440. void Acceptor::listen()  
1441. {  
1442.     loop_->assertInLoopThread();  
1443.     listening_ = true;  
1444.     acceptSocket_.listen();  
1445.     acceptChannel_.enableReading();  
1446. }  
1447. void Acceptor::handleRead()  
1448. {  
1449.     loop_->assertInLoopThread();  
1450.     InetAddress peerAddr(0);  
1451. int connfd = acceptSocket_.accept(&peerAddr);  
1452. if (connfd >= 0) {  
1453. if (newConnectionCallback_) {  
1454.             newConnectionCallback_(connfd, peerAddr);  
1455.         } else {  
1456.             sockets::close(connfd);  
1457.         }  
1458.     }  
1459. }  
1460. /*
1461.  *Buffer管理数据接收与发送
1462.  */
1463. class Buffer{//copyable
1464. public:  
1465. static const size_t kCheapPrepend=8;  
1466. static const size_t kInitialSize=1024;  
1467.         Buffer():buffer_(kCheapPrepend+kInitialSize),  
1468.             readerIndex_(kCheapPrepend),writerInex_(kCheapPrepend)  
1469.         {  
1470.             assert(readableBytes()==0);  
1471.             assert(writeableBytes()==kInitialSize);  
1472.             assert(prependableBytes()==kCheapPrepend);  
1473.         }  
1474. void swap(Buffer& rhs){  
1475.             buffer_.swap(rhs.buffer_);  
1476.             std::swap(readerIndex_,rhs.readerIndex_);  
1477.             std::swap(writerInex_,rhs.writerInex_);  
1478.         }  
1479. size_t readableBytes() const{  
1480. return writerInex_-readerIndex_;  
1481.         }//返回Buffer中多少数据
1482. size_t writeableBytes() const{  
1483. return buffer_.size()-writerInex_;  
1484.         }//返回还有多少剩余空间
1485. size_t prependableBytes() const{  
1486. return readerIndex_;  
1487.         }//返回可读位置
1488. const char* peek() const{  
1489. return begin()+readerIndex_;  
1490.         }//第一个可读的字节处
1491. void retrieve(size_t len){  
1492.             assert(len<=readableBytes());  
1493.             readerIndex_+=len;  
1494.         }//一次性没有读完，readerindex_移动
1495. void retrieveUntil(const char* end){  
1496.             assert(peek()<=end);  
1497.             assert(end<=beginWrite());//beginwrite()返回第一个可写的位置
1498.             retrieve(end-peek());  
1499.         }//返回有多少Buffer中可读字节
1500. void retrieveAll(){  
1501.             readerIndex_=kCheapPrepend;  
1502.             writerInex_=kCheapPrepend;  
1503.         }//重置Buffer
1504.         std::string retrieveAsString(){  
1505.             string str(peek(),readableBytes());  
1506.             retrieveAll();  
1507. return str;  
1508.         }//以string返回Buffer中数据，并重置Buffer
1509. void append(const string& str){  
1510.             append(str.data(),str.length());  
1511.         }  
1512. void append(const char* data,size_t len){  
1513.             ensureWriteableBytes(len);//空间不足会调用makespace扩容或者内部腾挪
1514.             std::copy(data,data+len,beginWrite());//copy(Input first,Input last,Output)
1515.             hasWritten(len);//更新writerinex_
1516.         }  
1517. void append(const void* data,size_t len){  
1518.             append(static_cast<const char*>(data),len);  
1519.         }  
1520. void ensureWriteableBytes(size_t len){  
1521. if(writeableBytes()<len){  
1522.                 makeSpace(len);  
1523.             }//若剩余空间不够，则重新分配空间
1524.             assert(writeableBytes()>=len);  
1525.         }  
1526. char* beginWrite(){  
1527. return begin()+writerInex_;  
1528.         }//可以写的位置
1529. const char* beginWrite() const{  
1530. return begin()+writerInex_;  
1531.         }  
1532. void hasWritten(size_t len){  
1533.             writerInex_+=len;  
1534.         }//更新writerinex_
1535. void prepend(const void* data,size_t len){  
1536.             assert(len<=prependableBytes());  
1537.             readerIndex_-=len;  
1538. const char* d=static_cast<const char*>(data);  
1539.             std::copy(d,d+len,begin()+readerIndex_);  
1540.         }//前向添加数据
1541. void shrink(size_t reserve){  
1542.             vector<char> buf(kCheapPrepend+readableBytes()+reserve);  
1543.             std::copy(peek(),peek()+readableBytes(),buf.begin()+kCheapPrepend);  
1544.             buf.swap(buffer_);  
1545.         }//重置Buffer大小
1546.         ssize_t readFd(int fd,int* savedErrno){  
1547. char extrabuf[65536];//栈空间,vector在堆空间
1548. struct iovec vec[2];  
1549. const size_t writeable=writeableBytes();  
1550.             vec[0].iov_base=begin()+writerInex_;  
1551.             vec[0].iov_len=writeable;  
1552.             vec[1].iov_base=extrabuf;  
1553.             vec[1].iov_len=sizeof extrabuf;  
1554. const ssize_t n=readv(fd,vec,2);//readv集中读
1555. if(n<0){  
1556.                 *savedErrno=errno;  
1557.             }  
1558. else if(implicit_cast<size_t>(n)<=writeable){  
1559.                 writerInex_+=n;  
1560.             }//Buffer还有剩余
1561. else{  
1562.                 writerInex_=buffer_.size();  
1563.                 append(extrabuf,n-writeable);  
1564.             }//Buffer不够，栈空间数据append到Buffer使Buffer慢慢变大
1565. return n;  
1566.         }  
1567. private:  
1568. char* begin(){//.>*>&首字符
1569. return &*buffer_.begin();  
1570.         }  
1571. const char* begin() const{  
1572. return &*buffer_.begin();  
1573.         }  
1574. void makeSpace(size_t len){//ensurewriteablebytes()->makespace()
1575. //当剩余空间writeable()<len时被调用
1576. if(writeableBytes()+prependableBytes()<len+kCheapPrepend){  
1577.                 buffer_.resize(writerInex_+len);  
1578.             }//(Buffer.size()-writerinex_剩余空间)+(readerindex_第一个可读位置)<
1579. //len+前向大小，这时无论怎样腾挪都不够写了，需要追加Buffer的大小
1580. else{//可以通过腾挪满足len大小的写入
1581.                 assert(kCheapPrepend<readerIndex_);  
1582. size_t readable=readableBytes();  
1583.                 std::copy(begin()+readerIndex_,begin()+writerInex_,begin()+kCheapPrepend);//Buffer的已有数据向前腾挪
1584.                 readerIndex_=kCheapPrepend;//readerindex_回到初始位置
1585.                 writerInex_=readerIndex_+readable;  
1586.                 assert(readable==readableBytes());  
1587.             }  
1588.         }  
1589. private:  
1590.         vector<char> buffer_;  
1591. size_t readerIndex_;  
1592. size_t writerInex_;  
1593. };  
1595. class TcpConnection;//表示一个TCP连接
1596. typedef shared_ptr<TcpConnection> TcpConnectionPtr;//
1597. /*
1598.  *TcpConnection
1599.  */
1600. class TcpConnection:noncopyable,public enable_shared_from_this<TcpConnection>{  
1601. public:  
1602.         TcpConnection(EventLoop* loop,const string& name,int sockfd,  
1603. const InetAddress& localAddr,const InetAddress& peerAddr);  
1604.         ~TcpConnection();  
1605.         EventLoop* getLoop() const{return loop_;}  
1606. const string& name() const{return name_;}  
1607. const InetAddress& localAddr(){return localAddr_;}  
1608. const InetAddress& peerAddress(){return peerAddr_;}  
1609. bool connected() const{return state_==kConnected;}  
1610. void send(const string& message);//发送消息,为了线程安全其会调用Tcpconnection::sendInLoop()
1611. void shutdown();//关闭TCP连接,为了保证线程安全其会调用Tcpconnection:shutdownInloop()
1612. void setTcpNoDelay(bool on);//关闭Nagle算法
1613. void setConnectionCallback(const ConnectionCallback& cb){  
1614.             connectionCallback_=cb;  
1615.         }//set*Callback系列函数是由用户通过Tcpserver::set*Callback指定并由TcpServer::newConnection()创建Tcpconnection对象时传递给Tcpconnection::set*Callback函数
1616. void setMessageCallback(const MessageCallback& cb){  
1617.             messageCallback_=cb;  
1618.         }  
1619. void setWriteCompleteCallback(const WriteCompleteCallback& cb){  
1620.             writeCompleteCallback_=cb;  
1621.         }  
1622. void setCloseCallback(const CloseCallback& cb){  
1623. //由TcpServer和TcpClient调用，解除它们中的TcpConnectionPtr
1624.             closeCallback_=cb;  
1625.         }  
1626. void connectEstablished();//调用Channel::enableReading()向Poller事件表注册事件，并调用TcpConnection::connectionCallback_()完成用户指定的连接回调
1627. //Acceptor::handleRead()->TcpServer::newConnection()->TcpConnection::connectEstablished()
1628. void connectDestroyed();//连接销毁函数,调用Channel::diableAll()使Poller对sockfd_忽略，并调用Eventloop::removeChannel()移除sockfd_对应的Channel
1629. //TcpServer::removeConnection()->EventLoop::runInLoop()->TcpServer::removeConnectionInLoop()->EventLoop::queueInLoop()->TcpConnection::connectDestroyed()
1630. //这是TcpConenction析构前调用的最后一个函数，用于告诉用户连接已断开
1631. //将TcpConenction状态置为kDisconnected,Channel::diableAll(),connectioncallback_(),EventLoop::removeChannel()
1632. private:  
1633. enum StateE{kConnecting,kConnected,kDisconnecting,kDisconnected,};  
1634. //Tcpconnection有四个状态:正在连接，已连接，正在断开，已断开
1635. void setState(StateE s){state_=s;}  
1636. void handleRead();  
1637. //Tcpconnection::handle*系列函数是由Poller返回sockfd_上就绪事件后由Channel::handelEvent()调用的就绪事件回调函数
1638. void handleWrite();  
1639. void handleClose();  
1640. void handleError();  
1641. void sendInLoop(const string& message);  
1642. void shutdownInLoop();  
1643.         EventLoop* loop_;  
1644.         string name_;  
1645.         StateE state_;  
1646.         scoped_ptr<Socket> socket_;//TcpConnection对应的那个TCP客户连接封装为socket_
1647.         scoped_ptr<Channel> channel_;//TcpConnection对应的TCP客户连接connfd采用Channel管理
1648.         InetAddress localAddr_;//TCP连接对应的服务端地址
1649.         InetAddress peerAddr_;//TCP连接的客户端地址
1650.         ConnectionCallback connectionCallback_;  
1651. //用户指定的连接回调函数,TcpServer::setConenctionCallback()->Acceptor::handleRead()->Tcpserver::newConnection()->TcpConnection::setConnectionCallback()
1652. //即TcpServer::setConenctionCallback()接收用户注册的连接回调，并通过Acceptor::handleRead()->Tcpserver::newConnection()将此用户连接回调函数传给Tcpconnection
1653.         MessageCallback messageCallback_;//用户指定的消息处理函数，也是经由Tcpserver传给Tcpconnection
1654.         WriteCompleteCallback writeCompleteCallback_;  
1655.         CloseCallback closeCallback_;  
1656.         Buffer inputBuffer_;  
1657.         Buffer outputBuffer_;  
1658. };  
1659. /*
1660.  *Tcpserver
1661.  */
1662. class TcpServer:noncopyable{//管理所有的TCP连接
1663. public:  
1664.         TcpServer(EventLoop* loop,const InetAddress& listenAddr);//构造时就有个监听端口的地址
1665.         ~TcpServer();  
1666. void setThreadNum(int numThreads);  
1667. void start();  
1668. void setConnectionCallback(const ConnectionCallback& cb){  
1669.             connectionCallback_=cb;  
1670.         }//TCP客户连接回调在TcpConnection里，TcpConnection::connectEstablished()->TcpConnection::connectionCallback_()
1671. void setMessageCallback(const MessageCallback& cb){  
1672.             messageCallback_=cb;  
1673.         }//此回调将传给TcpConnection::setMessageCallback()作为TcpConenction的消息回调
1674. void setWriteCompleteCallback(const WriteCompleteCallback& cb){  
1675.             writeCompleteCallback_=cb;  
1676.         }  
1677. private:  
1678. void newConnection(int sockfd,const InetAddress& peerAddr);  
1679. void removeConnection(const TcpConnectionPtr& conn);  
1680. void removeConnectionInLoop(const TcpConnectionPtr& conn);  
1681. typedef map<string,TcpConnectionPtr> ConnectionMap;  
1682.         EventLoop* loop_;  
1683. const string name_;  
1684.         scoped_ptr<Acceptor> acceptor_;//监听端口接受连接
1685.         scoped_ptr<EventLoopThreadPool> threadPool_;//开启EventLoopThreadPool管理TCP连接
1686.         ConnectionCallback connectionCallback_;//传给TcpConnection::setConnectionCallback(connectioncallback_)
1687.         MessageCallback messageCallback_;//传给TcpConnection::setMessageCallback(messagecallback_)
1688.         WriteCompleteCallback writeCompleteCallback_;  
1689. bool started_;  
1690. int nextConnId_;//用于标识TcpConnection，name_+nextConnId_就构成了一个TcpConnection的名字
1691.         ConnectionMap connections_;//该TcpServer管理的所有TCP客户连接存放的容器
1692. };  
1693. /*
1694.  *TcpConnection实现
1695.  */
1696. TcpConnection::TcpConnection(EventLoop* loop,  
1697. const std::string& nameArg,  
1698. int sockfd,  
1699. const InetAddress& localAddr,  
1700. const InetAddress& peerAddr)  
1701.     : loop_(loop),  
1702.     name_(nameArg),  
1703.     state_(kConnecting),  
1704.     socket_(new Socket(sockfd)),  
1705.     channel_(new Channel(loop, sockfd)),  
1706.     localAddr_(localAddr),  
1707.     peerAddr_(peerAddr)  
1708. {  
1709.     channel_->setReadCallback(bind(&TcpConnection::handleRead, this));  
1710.     channel_->setWriteCallback(bind(&TcpConnection::handleWrite,this));  
1711.     channel_->setCloseCallback(bind(&TcpConnection::handleClose,this));  
1712.     channel_->setErrorCallback(bind(&TcpConnection::handleError,this));  
1713. }  
1714. TcpConnection::~TcpConnection()  
1715. {  
1716.     printf("TcpConnection::%s,fd=%d\n",name_.c_str(),channel_->fd());  
1717. }  
1718. void TcpConnection::send(const string& message){  
1719.     cout<<"TcpConnection::send() ##"<<message<<endl;  
1720. if(state_==kConnected){  
1721. if(loop_->isInLoopThread()){  
1722.             sendInLoop(message);  
1723.         }  
1724. else{  
1725.             loop_->runInLoop(bind(&TcpConnection::sendInLoop,this,message));  
1726.         }  
1727.     }  
1728. }  
1729. void TcpConnection::sendInLoop(const string& message){  
1730. //若TcpConnection的socket已经注册了可写事件即outputBuffer_已经有数据了则直接调用Buffer::append
1731. //若socket的Channel没有注册可读则表明outputbuffer_没有数据存留，则可以直接先write发送
1732. //若write一次性没有发送完，则剩下数据需要append到outputbuffer_
1733. //若write一次性发送完毕则需要执行writecompletecallback_
1734.     loop_->assertInLoopThread();  
1735.     ssize_t nwrote=0;  
1736.     cout<<message<<endl;  
1737. if(!channel_->isWriting()&&outputBuffer_.readableBytes()==0){  
1738.         nwrote=write(channel_->fd(),message.data(),message.size());  
1739. if(nwrote>=0){  
1740. if(implicit_cast<size_t>(nwrote)<message.size()){  
1741.                 printf("I am going to write more data\n");  
1742.             }  
1743. else if(writeCompleteCallback_){  
1744.                 loop_->queueInLoop(bind(writeCompleteCallback_,shared_from_this()));  
1745.             }  
1746.         }  
1747. else{  
1748.             nwrote=0;  
1749. if(errno!=EWOULDBLOCK){  
1750.                 printf("TcpConnection::sendInLoop() error\n");  
1751.             }  
1752.         }  
1753.     }  
1754.     assert(nwrote>=0);  
1755. if(implicit_cast<size_t>(nwrote)<message.size()){  
1756.         outputBuffer_.append(message.data()+nwrote,message.size()-nwrote);  
1757. if(!channel_->isWriting()){  
1758.            channel_->enableWriting();  
1759.         }  
1760.     }  
1761. }  
1762. void TcpConnection::shutdown(){  
1763. if(state_==kConnected){  
1764.         setState(kDisconnecting);  
1765.         loop_->runInLoop(bind(&TcpConnection::shutdownInLoop,this));  
1766.     }  
1767. }  
1768. void TcpConnection::shutdownInLoop(){  
1769.     loop_->assertInLoopThread();  
1770. if(!channel_->isWriting()){  
1771.         socket_->shutdownWrite();  
1772.     }  
1773. }  
1774. void TcpConnection::setTcpNoDelay(bool on){  
1775.     socket_->setTcpNoDelay(on);  
1776. }  
1777. void TcpConnection::connectEstablished()  
1778. {  
1779.     loop_->assertInLoopThread();  
1780.     assert(state_ == kConnecting);  
1781.     setState(kConnected);  
1782.     channel_->enableReading();  
1783.     connectionCallback_(shared_from_this());//连接建立回调函数
1784. }  
1785. void TcpConnection::handleRead()  
1786. {  
1787. int savedErrno=0;  
1788.     ssize_t n =inputBuffer_.readFd(channel_->fd(),&savedErrno);//readv()
1789. if(n>0)  
1790.         messageCallback_(shared_from_this(),&inputBuffer_);  
1791. else if(n==0)  
1792.         handleClose();  
1793. else{  
1794.         errno=savedErrno;  
1795.         printf("TcpConnection::hanleRead() error\n");  
1796.         handleError();  
1797.     }  
1798. }  
1799. void TcpConnection::handleWrite(){  
1800.     loop_->assertInLoopThread();  
1801. if(channel_->isWriting()){  
1802.         ssize_t n=write(channel_->fd(),outputBuffer_.peek(),outputBuffer_.readableBytes());  
1803. if(n>0){//peek()返回第一个可读的字节，readablebytes()返回Buffer中数据的大小
1804.             outputBuffer_.retrieve(n);//readerindex_+=n更新Buffer的读位置
1805. if(outputBuffer_.readableBytes()==0){//如果Buffer里还有数据未发送的话不会立即调用shutdownwrite而是等待数据发送完毕再shutdown
1806.                 channel_->disableWriting();//防止busy loop
1807. if(writeCompleteCallback_){  
1808.                     loop_->queueInLoop(bind(writeCompleteCallback_,shared_from_this()));  
1809.                 }  
1810. if(state_==kDisconnecting)  
1811.                     shutdownInLoop();  
1812.             }  
1813. else
1814.                 printf("I am going to write more data\n");  
1815.         }  
1816. else
1817.             printf("TcpConnection::handleWrite()\n");  
1818.     }  
1819. else
1820.         printf("Connection is down,no more writing\n");  
1821. }  
1822. void TcpConnection::handleClose(){  
1823.     loop_->assertInLoopThread();  
1824.     assert(state_==kConnected||state_==kDisconnecting);  
1825.     channel_->disableAll();  
1826.     closeCallback_(shared_from_this());  
1827. }  
1828. void TcpConnection::handleError(){  
1829. int err=sockets::getSocketError(channel_->fd());  
1830.     printf("TcpConnection::handleError() %d %s\n",err,strerror(err));  
1831. }  
1832. void TcpConnection::connectDestroyed(){  
1833.     loop_->assertInLoopThread();  
1834.     printf("TcpConnection::handleClose() state=%s\n",state_);  
1835.     assert(state_==kConnected||state_==kDisconnected);  
1836.     setState(kDisconnected);  
1837.     channel_->disableAll();  
1838.     connectionCallback_(shared_from_this());  
1839.     loop_->removeChannel(get_pointer(channel_));  
1840. }  
1841. /*
1842.  *TcpServer实现
1843.  */
1844. TcpServer::TcpServer(EventLoop* loop, const InetAddressEventLoop::removeChannel()  
1845.     name_(listenAddr.toHostPort()),  
1846.     acceptor_(new Acceptor(loop, listenAddr)),  
1847.     threadPool_(new EventLoopThreadPool(loop)),  
1848.     started_(false),  
1849.     nextConnId_(1))  
1850. {  
1851.     acceptor_->setNewConnectionCallback(bind(&TcpServer::newConnection, this, _1, _2));  
1852. }  
1853. TcpServer::~TcpServer()  
1854. {  
1855. }  
1856. void TcpServer::setThreadNum(int numThreads){  
1857.     assert(numThreads>=0);  
1858.     threadPool_->setThreadNum(numThreads);  
1859. }  
1860. void TcpServer::start()  
1861. {  
1862. if (!started_)  
1863.     {  
1864.         started_ = true;  
1865.     }  
1867. if (!acceptor_->listening())  
1868.     {  
1869.         loop_->runInLoop(bind(&Acceptor::listen, get_pointer(acceptor_)));  
1870.     }//通过EventLoop监听服务端的listenfd,shared_ptr.hpp中的get_pointer用于返回shared_ptr所管理对象的裸指针
1871. }  
1872. void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)  
1873. {//用于Acceptor接受一个连接后通过此回调通知使用者
1874.     loop_->assertInLoopThread();  
1875. char buf[32];  
1876.     snprintf(buf, sizeof buf, "#%d", nextConnId_);  
1877.     ++nextConnId_;  
1878.     string connName = name_ + buf;  
1879.     InetAddress localAddr(sockets::getLocalAddr(sockfd));  
1880.     EventLoop* ioLoop=threadPool_->getNextLoop();//选一个EventLoop给TcpConnection
1881.     TcpConnectionPtr conn(  
1882. new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr));  
1883.     connections_[connName]=conn;  
1884.     conn->setConnectionCallback(connectionCallback_);//传递给TcpConnection
1885.     conn->setMessageCallback(messageCallback_);  
1886.     conn->setWriteCompleteCallback(writeCompleteCallback_);  
1887.     conn->setCloseCallback(bind(&TcpServer::removeConnection,this,_1));//将移除TcpConnectionPtr的操作注册到TcpConnection::setCloseCallback
1888.     ioLoop->runInLoop(bind(&TcpConnection::connectEstablished,conn));  
1889. //通过EventLoop::runInLoop()->EventLoop::queueInLoop()->TcpConnection::connectEstablished()
1890. }  
1891. void TcpServer::removeConnection(const TcpConnectionPtr& conn){  
1892.     loop_->runInLoop(bind(&TcpServer::removeConnectionInLoop,this,conn));  
1893. //TcpServer::removeConnection()->EventLoop::runInLoop()->EventLoop::queueInLoop()->TcpServer::removeConnectionInLoop()
1894. }  
1895. void TcpServer::removeConnectionInLoop(const TcpConnectionPtr& conn){  
1896.     loop_->assertInLoopThread();  
1897. size_t n=connections_.erase(conn->name());  
1898.     assert(n==1);  
1899.     (void)n;  
1900.     EventLoop* ioLoop=conn->getLoop();  
1901.     ioLoop->queueInLoop(bind(&TcpConnection::connectDestroyed,conn));//在IO线程内完成直接EventLoop::queueInLoop()
1902. }  
1903. /*
1904.  * 发起连接
1905.  */
1906. class Connector : boost::noncopyable  
1907. {  
1908. public:  
1909. typedef function<void (int sockfd)> NewConnectionCallback;  
1910.         Connector(EventLoop* loop, const InetAddress& serverAddr);  
1911.         ~Connector();  
1912. void setNewConnectionCallback(const NewConnectionCallback& cb)  
1913.         { newConnectionCallback_ = cb; }  
1914. void start();  // can be called in any thread
1915. void restart();  // must be called in loop thread
1916. void stop();  // can be called in any thread
1917. const InetAddress& serverAddress() const { return serverAddr_; }  
1918. private:  
1919. enum States { kDisconnected, kConnecting, kConnected };  
1920. //未连接，正在连接，已连接
1921. static const int kMaxRetryDelayMs = 30*1000;  
1922. static const int kInitRetryDelayMs = 500;  
1923. void setState(States s) { state_ = s; }  
1924. void startInLoop();  
1925. void connect();  
1926. void connecting(int sockfd);  
1927. void handleWrite();  
1928. void handleError();  
1929. void retry(int sockfd);  
1930. int removeAndResetChannel();  
1931. void resetChannel();  
1932.         EventLoop* loop_;  
1933.         InetAddress serverAddr_;  
1934. bool connect_; // atomic
1935.         States state_;  // FIXME: use atomic variable
1936.         boost::scoped_ptr<Channel> channel_;  
1937.         NewConnectionCallback newConnectionCallback_;  
1938. int retryDelayMs_;  
1939.         TimerId timerId_;  
1940. };  
1941. /*
1942.  * Connector实现
1943.  */
1944. typedef boost::shared_ptr<Connector> ConnectorPtr;  
1945. const int Connector::kMaxRetryDelayMs;  
1946. Connector::Connector(EventLoop* loop, const InetAddress& serverAddr)  
1947.     :loop_(loop),  
1948.     serverAddr_(serverAddr),  
1949.     connect_(false),  
1950.     state_(kDisconnected),  
1951.     retryDelayMs_(kInitRetryDelayMs)  
1952. {  
1953. }  
1954. Connector::~Connector()  
1955. {  
1956.     loop_->cancel(timerId_);  
1957.     assert(!channel_);  
1958. }  
1959. void Connector::start()  
1960. {//可以由其它线程调用
1961.     connect_ = true;  
1962.     loop_->runInLoop(boost::bind(&Connector::startInLoop, this)); // FIXME: unsafe
1963. }  
1964. void Connector::startInLoop()  
1965. {  
1966.     loop_->assertInLoopThread();  
1967.     assert(state_ == kDisconnected);  
1968. if (connect_)  
1969.     {  
1970.         connect();//
1971.     }  
1972. else
1973.     {}  
1974. }  
1975. void Connector::connect()  
1976. {  
1977. int sockfd = sockets::createNonblockingOrDie();  
1978. int ret = sockets::connect(sockfd, serverAddr_.getSockAddrInet());  
1979. int savedErrno = (ret == 0) ? 0 : errno;  
1980. switch (savedErrno)  
1981.     {  
1982. case 0:  
1983. case EINPROGRESS:  
1984. case EINTR:  
1985. case EISCONN:  
1986.             connecting(sockfd);  
1987. break;  
1989. case EAGAIN:  
1990. case EADDRINUSE:  
1991. case EADDRNOTAVAIL:  
1992. case ECONNREFUSED:  
1993. case ENETUNREACH:  
1994.             retry(sockfd);  
1995. break;  
1997. case EACCES:  
1998. case EPERM:  
1999. case EAFNOSUPPORT:  
2000. case EALREADY:  
2001. case EBADF:  
2002. case EFAULT:  
2003. case ENOTSOCK:  
2004.             sockets::close(sockfd);  
2005. break;  
2007. default:  
2008.             sockets::close(sockfd);  
2009. // connectErrorCallback_();
2010. break;  
2011.   }  
2012. }  
2013. void Connector::restart()  
2014. {  
2015.     loop_->assertInLoopThread();  
2016.     setState(kDisconnected);  
2017.     retryDelayMs_ = kInitRetryDelayMs;  
2018.     connect_ = true;  
2019.     startInLoop();  
2020. }  
2021. void Connector::stop()  
2022. {  
2023.     connect_ = false;  
2024.     loop_->cancel(timerId_);  
2025. }  
2026. void Connector::connecting(int sockfd)  
2027. {//EINPROGRESS
2028.     setState(kConnecting);  
2029.     assert(!channel_);  
2030.     channel_.reset(new Channel(loop_, sockfd));  
2031.     channel_->setWriteCallback(bind(&Connector::handleWrite, this)); // FIXME: unsafe
2032.     channel_->setErrorCallback(bind(&Connector::handleError, this)); // FIXME: unsafe
2033.     channel_->enableWriting();  
2034. }  
2035. int Connector::removeAndResetChannel()  
2036. {  
2037.     channel_->disableAll();  
2038.     loop_->removeChannel(get_pointer(channel_));  
2039. int sockfd = channel_->fd();  
2040.     loop_->queueInLoop(bind(&Connector::resetChannel, this)); // FIXME: unsafe
2041. return sockfd;  
2042. }  
2043. void Connector::resetChannel()  
2044. {  
2045.     channel_.reset();  
2046. }  
2047. void Connector::handleWrite()  
2048. {  
2049. if (state_ == kConnecting)  
2050.     {  
2051. int sockfd = removeAndResetChannel();  
2052. int err = sockets::getSocketError(sockfd);  
2053. if (err)  
2054.             retry(sockfd);  
2055. else if (sockets::isSelfConnect(sockfd))  
2056.             retry(sockfd);  
2057. else
2058.         {  
2059.             setState(kConnected);  
2060. if (connect_)  
2061.                 newConnectionCallback_(sockfd);  
2062. else
2063.                 sockets::close(sockfd);  
2064.         }  
2065.     }  
2066. else
2067.     {  
2068.         assert(state_ == kDisconnected);  
2069.     }  
2070. }  
2072. void Connector::handleError()  
2073. {  
2074.     assert(state_ == kConnecting);  
2076. int sockfd = removeAndResetChannel();  
2077. int err = sockets::getSocketError(sockfd);  
2078.     retry(sockfd);  
2079. }  
2081. void Connector::retry(int sockfd)  
2082. {//EAGAIN
2083.     sockets::close(sockfd);  
2084.     setState(kDisconnected);  
2085. if (connect_){  
2086.         timerId_ = loop_->runAfter(retryDelayMs_/1000.0,  // FIXME: unsafe
2087.                                boost::bind(&Connector::startInLoop, this));  
2088.         retryDelayMs_ = std::min(retryDelayMs_ * 2, kMaxRetryDelayMs);  
2089.     }  
2090. else
2091.     {}  
2092. }  
2093. /*
2094.  * TcpClient
2095.  */
2096. typedef boost::shared_ptr<Connector> ConnectorPtr;  
2097. class TcpClient : boost::noncopyable  
2098. {  
2099. public:  
2100.         TcpClient(EventLoop* loop,  
2101. const InetAddress& serverAddr,  
2102. const string& name);  
2103.      ~TcpClient();  // force out-line dtor, for scoped_ptr members.
2104. void connect();  
2105. void disconnect();  
2106. void stop();  
2107.         TcpConnectionPtr connection() const
2108.         {  
2109.             MutexLockGuard lock(mutex_);  
2110. return connection_;  
2111.         }  
2113.         EventLoop* getLoop() const { return loop_; }  
2114. bool retry() const;  
2115. void enableRetry() { retry_ = true; }  
2116. void setConnectionCallback(const ConnectionCallback& cb)  
2117.         { connectionCallback_ = cb; }  
2118. void setMessageCallback(const MessageCallback& cb)  
2119.         { messageCallback_ = cb; }  
2120. void setWriteCompleteCallback(const WriteCompleteCallback& cb)  
2121.         { writeCompleteCallback_ = cb; }  
2122.         #ifdef __GXX_EXPERIMENTAL_CXX0X__
2123. void setConnectionCallback(ConnectionCallback&& cb)  
2124.         { connectionCallback_ = cb; }  
2125. void setMessageCallback(MessageCallback&& cb)  
2126.         { messageCallback_ = cb; }  
2127. void setWriteCompleteCallback(WriteCompleteCallback&& cb)  
2128.         { writeCompleteCallback_ = cb; }  
2129.         #endif
2130. private:  
2131. void newConnection(int sockfd);  
2132. void removeConnection(const TcpConnectionPtr& conn);  
2133.         EventLoop* loop_;  
2134.         ConnectorPtr connector_; // avoid revealing Connector
2135. const string name_;  
2136.         ConnectionCallback connectionCallback_;  
2137.         MessageCallback messageCallback_;  
2138.         WriteCompleteCallback writeCompleteCallback_;  
2139. bool retry_;   // atmoic
2140. bool connect_; // atomic
2141. int nextConnId_;  
2142. mutable MutexLock mutex_;  
2143.         TcpConnectionPtr connection_; // @BuardedBy mutex_
2144. };  
2145. namespace detail  
2146. {  
2147. void removeConnection(EventLoop* loop, const TcpConnectionPtr& conn)  
2148.     {  
2149.       loop->queueInLoop(boost::bind(&TcpConnection::connectDestroyed, conn));  
2150.     }  
2151. void removeConnector(const ConnectorPtr& connector)  
2152.     {  
2153. //connector->
2154.     }  
2155. }  
2156. TcpClient::TcpClient(EventLoop* loop,  
2157. const InetAddress& serverAddr,  
2158. const string& name)  
2159.   : loop_(CHECK_NOTNULL(loop)),  
2160.     connector_(new Connector(loop, serverAddr)),  
2161.     name_(name),  
2162.     connectionCallback_(defaultConnectionCallback),  
2163.     messageCallback_(defaultMessageCallback),  
2164.     retry_(false),  
2165.     connect_(true),  
2166.     nextConnId_(1)  
2167. {  
2168.     connector_->setNewConnectionCallback(  
2169.       boost::bind(&TcpClient::newConnection, this, _1));  
2170. }  
2172. TcpClient::~TcpClient()  
2173. {  
2174.     TcpConnectionPtr conn;  
2175.     {  
2176.         MutexLockGuard lock(mutex_);  
2177.         conn = connection_;  
2178.     }  
2179. if (conn)  
2180.     {  
2181.         CloseCallback cb = boost::bind(&detail::removeConnection, loop_, _1);  
2182.         loop_->runInLoop(  
2183.             boost::bind(&TcpConnection::setCloseCallback, conn, cb));  
2184.     }  
2185. else
2186.     {  
2187.         connector_->stop();  
2188.         loop_->runAfter(1, boost::bind(&detail::removeConnector, connector_));  
2189.     }  
2190. }  
2191. void TcpClient::connect()  
2192. {  
2193.     connect_ = true;  
2194.     connector_->start();  
2195. }  
2196. void TcpClient::disconnect()  
2197. {  
2198.     connect_ = false;  
2199.     {  
2200.         MutexLockGuard lock(mutex_);  
2201. if (connection_)  
2202.         {  
2203.             connection_->shutdown();  
2204.         }  
2205.     }  
2206. }  
2207. void TcpClient::stop()  
2208. {  
2209.     connect_ = false;  
2210.     connector_->stop();  
2211. }  
2212. void TcpClient::newConnection(int sockfd)  
2213. {  
2214.     loop_->assertInLoopThread();  
2215.     InetAddress peerAddr(sockets::getPeerAddr(sockfd));  
2216. char buf[32];  
2217.     snprintf(buf, sizeof buf, ":%s#%d", peerAddr.toIpPort().c_str(), nextConnId_);  
2218.     ++nextConnId_;  
2219.     string connName = name_ + buf;  
2220.     InetAddress localAddr(sockets::getLocalAddr(sockfd));  
2221.     TcpConnectionPtr conn(new TcpConnection(loop_,  
2222.                                           connName,  
2223.                                           sockfd,  
2224.                                           localAddr,  
2225.                                           peerAddr));  
2227.     conn->setConnectionCallback(connectionCallback_);  
2228.     conn->setMessageCallback(messageCallback_);  
2229.     conn->setWriteCompleteCallback(writeCompleteCallback_);  
2230.     conn->setCloseCallback(  
2231.       boost::bind(&TcpClient::removeConnection, this, _1)); // FIXME: unsafe
2232.     {  
2233.         MutexLockGuard lock(mutex_);  
2234.         connection_ = conn;  
2235.     }  
2236.     conn->connectEstablished();  
2237. }  
2238. void TcpClient::removeConnection(const TcpConnectionPtr& conn)  
2239. {  
2240.     loop_->assertInLoopThread();  
2241.     assert(loop_ == conn->getLoop());  
2242.     {  
2243.         MutexLockGuard lock(mutex_);  
2244.         assert(connection_ == conn);  
2245.         connection_.reset();  
2246.     }  
2247.     loop_->queueInLoop(boost::bind(&TcpConnection::connectDestroyed, conn));  
2248. if (retry_ && connect_)  
2249.     {  
2250.         connector_->restart();  
2251.     }  
2252. }  
2254. /*
2255.  *Epoll
2256.  */
2257. class Epoller:noncopyable{  
2258. public:  
2259. typedef vector<Channel*> ChannelList;  
2260.         Epoller(EventLoop* loop)  
2261.             :ownerLoop_(loop),  
2262.             epollfd_(::epoll_create1(EPOLL_CLOEXEC)),  
2263.             events_(kInitEventListSize)  
2264.         {  
2265. if(epollfd_<0){  
2266.                 printf("Epoller::epoll_create1() error\n");  
2267.                 abort();  
2268.             }  
2269.         }  
2270.         ~Epoller(){  
2271.             ::close(epollfd_);  
2272.         }  
2273.         Timestamp poll(int timeoutMs,ChannelList* activeChannels){  
2274. int numEvents=::epoll_wait(epollfd_,&*events_.begin(),  
2275. static_cast<int>(events_.size()),timeoutMs);  
2276.             Timestamp now(Timestamp::now());  
2277. if(numEvents>0){  
2278.                 fillActiveChannels(numEvents,activeChannels);  
2279. if(implicit_cast<size_t>(numEvents)==events_.size()){  
2280.                     events_.resize(events_.size()*2);  
2281.                 }  
2282. else if(numEvents==0){}  
2283. else{  
2284.                     printf("Epoller::epoll_wait() error\n");  
2285.                 }  
2286.             }  
2287. return now;  
2288.         }  
2289. void updateChannel(Channel* channel){  
2290.             assertInLoopThread();  
2291. const int index=channel->index();  
2292. if(index==-1||index==2){  
2293. int fd=channel->fd();  
2294. if(index==-1){  
2295.                     assert(channels_.find(fd)==channels_.end());  
2296.                     channels_[fd]=channel;  
2297.                 }  
2298. else{  
2299.                     assert(channels_.find(fd)!=channels_.end());  
2300.                     assert(channels_[fd]==channel);  
2301.                 }  
2302.                 channel->set_index(1);  
2303.                 update(EPOLL_CTL_ADD,channel);  
2304.             }  
2305. else{  
2306. int fd=channel->fd();  
2307.                 (void)fd;  
2308.                 assert(channels_.find(fd)!=channels_.end());  
2309.                 assert(channels_[fd]==channel);  
2310.                 assert(index==1);  
2311. if(channel->isNoneEvent()){  
2312.                     update(EPOLL_CTL_DEL,channel);  
2313.                     channel->set_index(2);  
2314.                 }  
2315. else{  
2316.                     update(EPOLL_CTL_MOD,channel);  
2317.                 }  
2318.             }  
2319.         }  
2320. void removeChannel(Channel* channel){  
2321.             assertInLoopThread();  
2322. int fd=channel->fd();  
2323.             assert(channels_.find(fd)!=channels_.end());  
2324.             assert(channels_[fd]==channel);  
2325.             assert(channel->isNoneEvent());  
2326. int index=channel->index();  
2327.             assert(index==1||index==2);  
2328. size_t n=channels_.erase(fd);  
2329.             (void)n;  
2330.             assert(n==1);  
2331. if(index==1){  
2332.                 update(EPOLL_CTL_DEL,channel);  
2333.             }  
2334.             channel->set_index(-1);  
2335.         }  
2336. void assertInLoopThread(){  
2337.             ownerLoop_->assertInLoopThread();  
2338.         }  
2339. private:  
2340. static const int kInitEventListSize=16;  
2341. void fillActiveChannels(int numEvents,ChannelList* activeChannels) const
2342.         {  
2343.             assert(implicit_cast<size_t>(numEvents)<=events_.size());  
2344. for(int i=0;i<numEvents;i++){  
2345.                 Channel* channel=static_cast<Channel*>(events_[i].data.ptr);  
2346. int fd=channel->fd();  
2347.                 ChannelMap::const_iterator it=channels_.find(fd);  
2348.                 assert(it!=channels_.end());  
2349.                 assert(it->second==channel);  
2350.                 channel->set_revents(events_[i].events);  
2351.                 activeChannels->push_back(channel);  
2352.             }  
2353.         }  
2354. void update(int operation,Channel* channel){  
2355. struct epoll_event event;  
2356.             bzero(&event,sizeof event);  
2357.             event.events=channel->events();  
2358.             event.data.ptr=channel;  
2359. int fd=channel->fd();  
2360. if(::epoll_ctl(epollfd_,operation,fd,&event)<0){  
2361. if(operation==EPOLL_CTL_DEL){  
2362.                     printf("Epoller::update() EPOLL_CTL_DEL error\n");  
2363.                 }  
2364. else{  
2365.                     printf("Epoller::update() EPOLL_CTL_ error\n");  
2366.                 }  
2367.             }  
2368.         }  
2369. typedef vector<struct epoll_event> EventList;  
2370. typedef map<int,Channel*> ChannelMap;  
2372.         EventLoop* ownerLoop_;  
2373. int epollfd_;  
2374.         EventList events_;  
2375.         ChannelMap channels_;  
2376. };