EventLoop中時(shí)序深究:
過(guò)程:
1.在程序中我們一般會(huì)先構(gòu)造一個(gè)EventLoop對(duì)象(muduo::EventLoop loop),EventLoop構(gòu)造函數(shù)初始化列表,構(gòu)造poller_,timerQueue_,wakeupFd_和wakeupChannel_等成員,在函數(shù)中具體表現(xiàn)為:
poller_(new Poller(this)),
timerQueue_(new TimerQueue(this)),
wakeupFd_(createEventfd()),
wakeupChannel_(new Channel(this, wakeupFd_))
其中wakeupFd_通過(guò)createEventfd()函數(shù)來(lái)初始化,在后者的函數(shù)中通過(guò)eventfd()實(shí)現(xiàn)線程高效的喚醒,wakeupFd_將用于后面的wakeup()函數(shù),用來(lái)喚醒線程.
wakeupChannel_->setReadCallback(
boost::bind(&EventLoop::handleRead, this));
// we are always reading the wakeupfd
wakeupChannel_->enableReading();
對(duì)于wakeupChannel_,調(diào)用setReadCallback()函數(shù)注冊(cè)wakeupChannel回調(diào)函數(shù)為EventLoop::handleRead,后者會(huì)調(diào)用::read來(lái)read掉weakupFd_上的可讀事件,避免一直觸發(fā),后面會(huì)說(shuō).另外wakeupChannel會(huì)調(diào)用enableReading,后者接著會(huì)調(diào)用Channel::update();進(jìn)而調(diào)用EventLoop::UpdataChannel();最后調(diào)用Poller::updataChannel(),在此函數(shù)內(nèi)會(huì)添加一個(gè)新的Channel或者更新已經(jīng)存在的此channel關(guān)注的事件.在一開始會(huì)將wakeupChannel_添加進(jìn)Poller::channels_中,其類型為(typedef std::map<int, Channel*> ChannelMap;);并使用wakeupChannel_.fd和wakeupChannel_.events構(gòu)造一個(gè)struct pollfd pfd,然后將pfd push_back到Poller::pollfds_,以后會(huì)關(guān)注wakeupChannel_(wakeupFd_)上的可讀事件.
2.在程序中事件的循環(huán)開始于EventLoop::loop(),其內(nèi)部通過(guò)調(diào)用poller_->poll()(假設(shè)是采用的poll而不是epoll).::poll()函數(shù)會(huì)阻塞返回,即有事件發(fā)生,如socket上有數(shù)據(jù)可讀寫;Poller::poll中會(huì)調(diào)用fillActiveChannels()函數(shù),該函數(shù)中會(huì)將pollfds_中的pollfd結(jié)構(gòu)體中的revents設(shè)置為此channel的revents_,然后將次channel壓入activeChannels中,然后將其作為EventLoop::activeChannels_返回.接著在EventLoop::loop中遍歷activeChannels_,對(duì)每個(gè)channel調(diào)用Channel::handleEvent(),進(jìn)而調(diào)用每個(gè)channel注冊(cè)的讀寫回調(diào)函數(shù).
3.EventLoop::runInLoop(const Functor& cb)函數(shù)是為了在IO線程中執(zhí)行某個(gè)回調(diào)函數(shù),其主要功能是可以跨線程調(diào)用.EventLoop::doPendingFunctors(),用于處理被掛起的事件,該函數(shù)只會(huì)在當(dāng)前IO線程中調(diào)用.
測(cè)試程序
#include "EventLoop.h"
#include <stdio.h>
muduo::EventLoop* g_loop;
int g_flag = 0;
void run4()
{
printf("run4(): pid = %d, flag = %d\n", getpid(), g_flag);
g_loop->quit();
}
void run3()
{
printf("run3(): pid = %d, flag = %d\n", getpid(), g_flag);
g_loop->runAfter(5, run4);
g_flag = 3;
printf("%d\n",g_flag);
}
void run2()
{
printf("run2(): pid = %d, flag = %d\n", getpid(), g_flag);
g_loop->queueInLoop(run3);
}
void run1()
{
g_flag = 1;
printf("run1(): pid = %d, flag = %d\n", getpid(), g_flag);
g_loop->runInLoop(run2);
g_flag = 2;
}
int main()
{
printf("main(): pid = %d, flag = %d\n", getpid(), g_flag);
muduo::EventLoop loop;
g_loop = &loop;
loop.runAfter(2, run1);
loop.loop();
printf("main(): pid = %d, flag = %d\n", getpid(), g_flag);
}
執(zhí)行結(jié)果:
main(): pid = 28847, flag = 0
run1(): pid = 28847, flag = 1
run2(): pid = 28847, flag = 1
run3(): pid = 28847, flag = 2
3
run4(): pid = 28847, flag = 3
main(): pid = 28847, flag = 3
此為單線程程序,在loop內(nèi)定義了一個(gè)2秒之后執(zhí)行的回調(diào)函數(shù)run1(),在run1()內(nèi),由于就是在當(dāng)前IO線程,那么就會(huì)馬上執(zhí)行g(shù)_loop->runInLoop(run2),此時(shí)flag還是為1,接著在run2()中,執(zhí)行g(shù)_loop->queueInLoop(run3);即把run3()函數(shù)加入到隊(duì)列中,run2()返回,此時(shí)flag=2;此時(shí)主線程中的loop已經(jīng)處理完事件了,之后就處理執(zhí)行掛起事件,即執(zhí)行doPendingFunctors()第晰，就執(zhí)行了run3(), run3()內(nèi)設(shè)置另一個(gè)5s定時(shí)器脉漏，run3()執(zhí)行完回到loop繼續(xù)poll, 3s后超時(shí)執(zhí)行run4()，此時(shí)flag=3.
Acceptor類
Acceptor用于accept(2)新TCP連接,并通過(guò)回調(diào)通知使用者.它是內(nèi)部的class,供TcpServer使用,生命期由后者控制.
Acceptor的數(shù)據(jù)成員包括acceptSocket_采呐、acceptChannel_，Acceptor的acceptSocket_是正在監(jiān)聽的socket(即server socket).acceptChannel_是用于觀察acceptSocket_上的可讀事件,萬(wàn)一有可讀事件發(fā)生,那么Channel::handleEvent函數(shù)便會(huì)被調(diào)用,在其中回調(diào)Acceptor::handleRead().
acceptChannel_.setReadCallback( boost::bind(&Acceptor::handleRead, this)); //構(gòu)造函數(shù)中
然后在Acceptor::handleRead()函數(shù)中,使用accept(2)來(lái)接受新連接,并且可以回調(diào)用戶的callback,可以看到回調(diào)函數(shù)中的第一個(gè)參數(shù)為accept返回的connfd.
void setNewConnectionCallback(const NewConnectionCallback& cb)
{
newConnectionCallback_ = cb;
}
void Acceptor::handleRead()
{
loop_->assertInLoopThread();
InetAddress peerAddr(0);
//FIXME loop until no more
int connfd = acceptSocket_.accept(&peerAddr);
if (connfd >= 0) {
if (newConnectionCallback_) {
newConnectionCallback_(connfd, peerAddr);
} else {
sockets::close(connfd);
}
}
}
在該類中還有一個(gè)成員函數(shù):listen,其作用是用于監(jiān)聽.
void Acceptor::listen()
{
loop_->assertInLoopThread();
listenning_ = true;
acceptSocket_.listen();
acceptChannel_.enableReading();
}
TcpServer與TcpConnection:
在上層應(yīng)用程序中,我們一般不會(huì)直接調(diào)用Acceptor類,而是借助于TcpServer,把Acceptor作為數(shù)據(jù)成員使用.
該類的功能是管理accept(2)獲得TcpConnection,是給用戶直接使用的,生命期由用戶來(lái)控制.
TcpServer的Acceptor與TcpConnection數(shù)據(jù)成員:
typedef boost::shared_ptr<TcpConnection> TcpConnectionPtr;
typedef std::map<std::string, TcpConnectionPtr> ConnectionMap
boost::scoped_ptr<Acceptor> acceptor_; // avoid revealing Acceptor
ConnectionCallback connectionCallback_;
TcpConnection類使用Channel來(lái)獲得socket上的IO事件,他會(huì)自己處理可寫事件,而把可讀事件通過(guò)MessageCallback傳達(dá)給客戶,其沒(méi)有直接發(fā)起連接的功能,其構(gòu)造函數(shù)的參數(shù)是已經(jīng)建立好的socket.因此其中有兩個(gè)重要的數(shù)據(jù)成員:
boost::scoped_ptr<Socket> socket_;
boost::scoped_ptr<Channel> channel_;
時(shí)序分析
在程序中我們創(chuàng)建一個(gè)TcpServer對(duì)象,在其構(gòu)造函數(shù)中我們會(huì)初始化TcpServer::acceptor_對(duì)象,并且設(shè)置acceptor_中的handleRead中回調(diào)TcpServer::newConnection()函數(shù).
acceptor_(new Acceptor(loop, listenAddr));
//setNewConnectionCallback函數(shù)在Acceptor中用來(lái)設(shè)置newConnectionCallback_,而此時(shí)將它設(shè)置為TcpServer::newConnection.
acceptor_->setNewConnectionCallback(
boost::bind(&TcpServer::newConnection, this, _1, _2));
然后該TcpServer對(duì)象調(diào)用start(),在該函數(shù)中會(huì)調(diào)用Acceptor::listen()函數(shù)來(lái)啟動(dòng)監(jiān)聽;若有活動(dòng)連接,即TcpServer::acceptor_中的acceptChannel_ 對(duì)象可讀,則poll返回,調(diào)用Channel::handleEvent()來(lái)處理活動(dòng)通道,在該函數(shù)中又被注冊(cè)了Acceptor::handleRead()回調(diào)函數(shù),在該函數(shù)中,使用accept(2)來(lái)接受新連接,并且會(huì)回調(diào)TcpServer構(gòu)造函數(shù)中注冊(cè)的TcpServer::newConnection()函數(shù),在該函數(shù)中:
TcpConnectionPtr conn(
new TcpConnection(loop_, connName, sockfd, localAddr, peerAddr));
connections_[connName] = conn;
conn->setConnectionCallback(connectionCallback_);
conn->setMessageCallback(messageCallback_);
conn->setCloseCallback(
boost::bind(&TcpServer::removeConnection, this, _1));
conn->connectEstablished();
會(huì)創(chuàng)建一個(gè)新的TcpConnectionPtr變量conn,在該變量的構(gòu)造函數(shù)中我們?cè)O(shè)置了
// 通道可讀事件到來(lái)的時(shí)候，回調(diào)TcpConnection::handleRead，_1是事件發(fā)生時(shí)間
channel_->setReadCallback(
boost::bind(&TcpConnection::handleRead, this, _1));
并且在該函數(shù)中我們對(duì)conn設(shè)置了連接到來(lái)的回調(diào)函數(shù)和消息回調(diào)函數(shù)(這是在TcpServer構(gòu)造函數(shù)中初始化的).最后我們調(diào)用TcpConnection::connectEstablished(),會(huì)將該TcpConnection所對(duì)應(yīng)的通道加入到Poller關(guān)注.至此我們已經(jīng)成功建立了一個(gè)連接,當(dāng)對(duì)方發(fā)送數(shù)據(jù)到connfd,內(nèi)核接受緩沖區(qū)不在為空,觸發(fā)可讀事件,
TcpConnection::channel_ 可讀事件發(fā)生锨阿，poll返回，調(diào)用Channel::handleEvent()處理活動(dòng)通道记罚，調(diào)用TcpConnection::handleRead(),在handleRead()中會(huì)回調(diào)messageCallback_函數(shù).
void TcpConnection::handleRead(Timestamp receiveTime)
{
ssize_t n = ::read(channel_->fd(), buf, sizeof buf);
messageCallback_(shared_from_this(), buf, n);
}
測(cè)試程序
#include "TcpServer.h"
#include "EventLoop.h"
#include "InetAddress.h"
#include <stdio.h>
void onConnection(const muduo::TcpConnectionPtr& conn)
{
if (conn->connected())
{
printf("onConnection(): new connection [%s] from %s\n",
conn->name().c_str(),
conn->peerAddress().toHostPort().c_str());
}
else
{
printf("onConnection(): connection [%s] is down\n",
conn->name().c_str());
}
}
void onMessage(const muduo::TcpConnectionPtr& conn,
muduo::Buffer* buf,
muduo::Timestamp receiveTime)
{
printf("onMessage(): received %zd bytes from connection [%s] at %s\n",
buf->readableBytes(),
conn->name().c_str(),
receiveTime.toFormattedString().c_str());
printf("onMessage(): [%s]\n", buf->retrieveAsString().c_str());
}
int main()
{
printf("main(): pid = %d\n", getpid());
muduo::InetAddress listenAddr(9981);
muduo::EventLoop loop;
muduo::TcpServer server(&loop, listenAddr);
server.setConnectionCallback(onConnection);
server.setMessageCallback(onMessage);
server.start();
loop.loop();
}
運(yùn)行結(jié)果:
main(): pid = 5983
20191008 08:48:51.871004Z 5983 INFO TcpServer::newConnection [0.0.0.0:9981] - new connection [0.0.0.0:9981#1] from 127.0.0.1:41400 - TcpServer.cc:58
20191008 08:48:51.871146Z 5983 DEBUG TcpConnection TcpConnection::ctor[0.0.0.0:9981#1] at 0x10624a0 fd=14 - TcpConnection.cc:36
onConnection(): new connection [0.0.0.0:9981#1] from 127.0.0.1:41400
onMessage(): received 4 bytes from connection [0.0.0.0:9981#1] at 20191008 08:49:05.568592
onMessage(): [jij
]
20191008 08:49:09.272725Z 5983 INFO TcpServer::removeConnection [0.0.0.0:9981] - connection 0.0.0.0:9981#1 - TcpServer.cc:76
onConnection(): connection [0.0.0.0:9981#1] is down
20191008 08:49:09.272830Z 5983 DEBUG ~TcpConnection TcpConnection::dtor[0.0.0.0:9981#1] at 0x10624a0 fd=14 - TcpConnection.cc:50
