0 设计
EventLoop 类，主要是用来管理一个进程中的channel、计时器、以及epoll 的类。
每个类只有一个，（因为如果有两个，那么一个eventloop 在loop()的时候，另一个eventloop 得不到执行）。
每一个channel或者说是每一个文件描述符都必须属于eventloop，换句话说，每一个文件描述符属于一个线程。（这是必然的，因为eventloop在代表了一个线程）
eventloop 类的主要工作，就是wait epoll，然后执行channel 的事件处理函数。
1 源码
#ifndef MUDUO_NET_EVENTLOOP_H
#define MUDUO_NET_EVENTLOOP_H
#include <vector>
#include <boost/any.hpp>
#include <boost/function.hpp>
#include <boost/noncopyable.hpp>
#include <boost/scoped_ptr.hpp>
#include <muduo/base/Mutex.h>
#include <muduo/base/CurrentThread.h>
#include <muduo/base/Timestamp.h>
#include <muduo/net/Callbacks.h>
#include <muduo/net/TimerId.h>
namespace muduo
{
namespace net
{
class Channel;
class Poller;
class TimerQueue;
// 事件处理类
class EventLoop : boost::noncopyable
{
public:
typedef boost::function<void()> Functor;
EventLoop();
~EventLoop();
// 里面一个while循环,直接开始工作.
void loop();
// 别的线程调用eventloop 的quit,设置 loop() 中循环的判断条件
// 从而在下一次循环的时候停止
void quit();
// 返回上一次 epoll 的返回时间
Timestamp pollReturnTime() const { return pollReturnTime_; }
int64_t iteration() const { return iteration_; }
// 添加函数对象到eventloop线程中执行。
void runInLoop(const Functor &cb);
// 当调用函数添加函数对象到eventloop中执行的时候
// 需要判断是否是在eventloop 线程中.如果是在当前eventloop 的线程中
// 那么直接执行,否则,就添加到队列中,然后唤醒epoll
void queueInLoop(const Functor &cb);
// 返回当前 等待执行函数对象的队列长度
size_t queueSize() const;
// 下面三个函数,是用来添加计时器对象的
TimerId runAt(const Timestamp &time, const TimerCallback &cb);
TimerId runAfter(double delay, const TimerCallback &cb);
TimerId runEvery(double interval, const TimerCallback &cb);
void cancel(TimerId timerId);
// wakeup函数用来显示的唤醒 epoll ,主要是在 eventloop执行了quit
// 还有添加了函数对象到队列中的时候 主动的唤醒 epoll
// 能唤醒的原因在于,eventloop 对象持有一个 eventfd,注册在了epoll中
// 然后需要唤醒的时候,直接 eventfd 中写数据即可.
void wakeup();
// 这两个函数被channel 的update 和remove 调用
// 然后,这俩函数由去调用epoll 的对应接口
void updateChannel(Channel *channel);
void removeChannel(Channel *channel);
// 用来判断一个 epoll 是否持有 这个 channel
// 调用的是 epoll 的对象
bool hasChannel(Channel *channel);
// 用来判断 绑定在eventloop 中的channel 的运行是否在 本eventloop 线程中
// 或是其他操作,需要属于本eventloop 线程
void assertInLoopThread()
{
if (!isInLoopThread())
{
abortNotInLoopThread();
}
}
// 被上面的函数调用,就是简单的判断下初始化eventloop 的线程是不是当前运行函数的线程
bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }
// eventHandling() 用来判断是否是在已就绪channel的事件处理函数。
// 因为channel的处理只能是在 eventloop 线程中，也就是说在 loop 处理的时候
// 如果这个时候 remove channel 那么要删除的channel 就是当前正在处理channel
bool eventHandling() const { return eventHandling_; }
// 没被用到,貌似
void setContext(const boost::any &context)
{
context_ = context;
}
const boost::any &getContext() const
{
return context_;
}
boost::any *getMutableContext()
{
return &context_;
}
static EventLoop *getEventLoopOfCurrentThread();
private:
void abortNotInLoopThread();
// 这个函数是在,eventloop 主动唤醒 epoll 的时候处理eventfd 读事件的函数
void handleRead(); // waked up
// 执行队列中的函数对象
void doPendingFunctors();
// 用来打印本次已就绪channel 发生的对应的时间,也就是去调用了channel 的tostring
void printActiveChannels() const; // DEBUG
typedef std::vector<Channel *> ChannelList;
bool looping_;
bool quit_;
bool eventHandling_;
bool callingPendingFunctors_;
// iteration_ 是poller 被唤醒的次数
int64_t iteration_;
// 初始化时候,本eventloop 所在的线程
const pid_t threadId_;
// 上一次epoll被唤醒的时间戳
Timestamp pollReturnTime_;
// 具体的epoll ,实现上可是用poll的,因此是一个继承体系
boost::scoped_ptr<Poller> poller_;
// 计时器队列.
boost::scoped_ptr<TimerQueue> timerQueue_;
// eventfd，是eventloop主动唤醒epoll的fd
int wakeupFd_;
// 是分装eventfd的channel
boost::scoped_ptr<Channel> wakeupChannel_;
boost::any context_;
// 这个容器被epoll 填充。装了所有已就绪的channel
ChannelList activeChannels_;
// 当前正在处理的 channel
Channel *currentActiveChannel_;
mutable MutexLock mutex_;
// 添加到eventloop 中需要执行的函数对象，就在这里。
std::vector<Functor> pendingFunctors_;
};
}
}
#endif // MUDUO_NET_EVENTLOOP_H
实现
#include <muduo/net/EventLoop.h>
#include <muduo/base/Logging.h>
#include <muduo/base/Mutex.h>
#include <muduo/net/Channel.h>
#include <muduo/net/Poller.h>
#include <muduo/net/SocketsOps.h>
#include <muduo/net/TimerQueue.h>
#include <boost/bind.hpp>
#include <signal.h>
#include <sys/eventfd.h>
#include <unistd.h>
using namespace muduo;
using namespace muduo::net;
// 当定义一个命名空间时，可以忽略这个命名空间的名称：
// 编译器在内部会为这个命名空间生成一个唯一的名字，而且还会为这个匿名的命名空间生成一条using指令。
// 命名空间都是具有external 连接属性的,只是匿名的命名空间产生的__UNIQUE_NAME__在别的文件中无法得到,这个唯一的名字是不可见的.
namespace
{
// 使用了 __thread 每个线程保存自己的eventloop指针
__thread EventLoop *t_loopInThisThread = 0;
// epoll 的超时时间
const int kPollTimeMs = 10000;
// 是eventloop 的eventfd ，用来唤醒 epoll
// 唤醒的时机在前面说了
int createEventfd()
{
int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
if (evtfd < 0)
{
LOG_SYSERR << "Failed in eventfd";
// 如果创建失败,那么直接结束程序,没有必要执行下去了
abort();
}
return evtfd;
}
// 这是两条预处理执行令,表示发生 -Wold-style-cast 的时候忽略
// 然后到下一条.
#pragma GCC diagnostic ignored "-Wold-style-cast"
// 这个对象只是用一次。
// 是为了防止管道破裂的。因此在构造的时候，直接将SIGPIPE设置为忽略
// 并且初始化的时候就执行了
class IgnoreSigPipe
{
public:
IgnoreSigPipe()
{
::signal(SIGPIPE, SIG_IGN);
// LOG_TRACE << "Ignore SIGPIPE";
}
};
// 到这里的时候开启.
#pragma GCC diagnostic error "-Wold-style-cast"
IgnoreSigPipe initObj;
}
// 返回本线程的 eventloop 对象
EventLoop *EventLoop::getEventLoopOfCurrentThread()
{
return t_loopInThisThread;
}
EventLoop::EventLoop()
: looping_(false),
quit_(false),
eventHandling_(false),
callingPendingFunctors_(false),
iteration_(0),
threadId_(CurrentThread::tid()),
poller_(Poller::newDefaultPoller(this)), // 使用一个工厂模式返回poll
timerQueue_(new TimerQueue(this)), // 和计时器有关的
wakeupFd_(createEventfd()), // 创建eventloop 主动唤醒 epoll 的channel
wakeupChannel_(new Channel(this, wakeupFd_)),
currentActiveChannel_(NULL)
{
LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;
// 如果当前线程已经有一个eventloop 了那么就退出啊。 LOG_FATAL 应该会直接停了。
if (t_loopInThisThread)
{
LOG_FATAL << "Another EventLoop " << t_loopInThisThread
<< " exists in this thread " << threadId_;
}
else
{
t_loopInThisThread = this;
}
wakeupChannel_->setReadCallback(
boost::bind(&EventLoop::handleRead, this));
// we are always reading the wakeupfd
wakeupChannel_->enableReading();
}
// 析构函数
EventLoop::~EventLoop()
{
LOG_DEBUG << "EventLoop " << this << " of thread " << threadId_
<< " destructs in thread " << CurrentThread::tid();
// 先将 wakechannel 注销掉，然后直接关闭就行了。
// 因为也没什么需要回收的资源,都是使用指针指针.
wakeupChannel_->disableAll();
wakeupChannel_->remove();
::close(wakeupFd_);
t_loopInThisThread = NULL;
}
void EventLoop::loop()
{
assert(!looping_);
assertInLoopThread();
looping_ = true;
quit_ = false;
LOG_TRACE << "EventLoop " << this << " start looping";
// 使用 quit 判断是否应该继续循环
while (!quit_)
{
// activeChannels_ 用来填充就绪channel的
activeChannels_.clear();
// 传入了 activeChannels_ 的地址,由poll 填充已就绪的channel 到activeChannels_中
pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
// iteration_ 是poller 被唤醒的次数
++iteration_;
// 根据日志级别,决定是否打印出所有的channel 的事件
if (Logger::logLevel() <= Logger::TRACE)
{
printActiveChannels();
}
// 接下来遍历所有的 activeChannels_ 中的元素,去执行它们的
// handleEvent() 实际的执行channel 的事件处理函数
// 这里面可能有 TimerQueue 的channel ，用来处理已经超时的计时器
// 这些计时器，被 TimerQueue 通过 add 添加函数对象的形式添加到了队列中？或是直接执行了。
// 想一下，应该是直接执行了，因为是在本线程中。
// 嗯在for中直接执行了每一个超时的计时器
eventHandling_ = true;
for (ChannelList::iterator it = activeChannels_.begin();
it != activeChannels_.end(); ++it)
{
// 这里保存一次 currentActiveChannel_ 的意义是
// 只是在删除额时候判断，删除的是否是这个channel、
// 还有一个没看懂
currentActiveChannel_ = *it;
currentActiveChannel_->handleEvent(pollReturnTime_);
}
currentActiveChannel_ = NULL;
eventHandling_ = false;
// 执行add 到eventloop 的函数对象。
doPendingFunctors();
}
LOG_TRACE << "EventLoop " << this << " stop looping";
looping_ = false;
}
// 退出。
void EventLoop::quit()
{
quit_ = true;
// 推出的时候，直接wakeup 唤醒 epoll
if (!isInLoopThread())
{
wakeup();
}
}
void EventLoop::runInLoop(const Functor &cb)
{
// 添加函数对象，如果是在本线程中就直接执行，否则就添加到队列中
if (isInLoopThread())
{
cb();
}
else
{
queueInLoop(cb);
}
}
// 添加到队列中
void EventLoop::queueInLoop(const Functor &cb)
{
{
MutexLockGuard lock(mutex_);
pendingFunctors_.push_back(cb);
}
// 如果不是在本队列中，或是在执行添加的函数对象
// 那么本次添加的就已经不能在本次epoll唤醒的时候调用了
// 因此这里直接wakeup。那么epoll 的wait直接被唤醒了
if (!isInLoopThread() || callingPendingFunctors_)
{
wakeup();
}
}
size_t EventLoop::queueSize() const
{
MutexLockGuard lock(mutex_);
return pendingFunctors_.size();
}
// 下面三个函数用来添加计时器的
TimerId EventLoop::runAt(const Timestamp &time, const TimerCallback &cb)
{
return timerQueue_->addTimer(cb, time, 0.0);
}
TimerId EventLoop::runAfter(double delay, const TimerCallback &cb)
{
Timestamp time(addTime(Timestamp::now(), delay));
return runAt(time, cb);
}
TimerId EventLoop::runEvery(double interval, const TimerCallback &cb)
{
Timestamp time(addTime(Timestamp::now(), interval));
return timerQueue_->addTimer(cb, time, interval);
}
// 取消计时器。
void EventLoop::cancel(TimerId timerId)
{
return timerQueue_->cancel(timerId);
}
// channel 调用的updatechannel，被转发了
void EventLoop::updateChannel(Channel *channel)
{
assert(channel->ownerLoop() == this);
assertInLoopThread();
poller_->updateChannel(channel);
}
// 同上
void EventLoop::removeChannel(Channel *channel)
{
assert(channel->ownerLoop() == this);
assertInLoopThread();
// 这里使用了 eventHandling_ ，如果是正在处理channel 的事件函数，
// 此时又要移除channel，那么要溢出的channel 必须是本channel 或是不在 activechannels中的channel
// 否则报错.这里为什么要这样处理? 一般来说都是本channel移除自己.
if (eventHandling_)
{
assert(currentActiveChannel_ == channel ||
std::find(activeChannels_.begin(), activeChannels_.end(), channel) == activeChannels_.end());
}
poller_->removeChannel(channel);
}
bool EventLoop::hasChannel(Channel *channel)
{
assert(channel->ownerLoop() == this);
assertInLoopThread();
return poller_->hasChannel(channel);
}
// 当不是在eventloop 的线程执行 eventloop 的相关操作,就会调用该函数.end
// 打印日志，然后退出
void EventLoop::abortNotInLoopThread()
{
LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
<< " was created in threadId_ = " << threadId_
<< ", current thread id = " << CurrentThread::tid();
}
// 主动唤醒的函数，向eventfd 中写入数据。
void EventLoop::wakeup()
{
uint64_t one = 1;
ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);
if (n != sizeof one)
{
LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
}
}
// 主动唤醒epoll 后，需要处理eventfd 的读事件，因此读出来就行了
void EventLoop::handleRead()
{
uint64_t one = 1;
ssize_t n = sockets::read(wakeupFd_, &one, sizeof one);
if (n != sizeof one)
{
LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
}
}
// 执行添加到eventloop 的函数对象。
// 这里没有使用拷贝，而是通过交换 vector 的方式。
// 更加效率
void EventLoop::doPendingFunctors()
{
std::vector<Functor> functors;
callingPendingFunctors_ = true;
{
MutexLockGuard lock(mutex_);
functors.swap(pendingFunctors_);
}
for (size_t i = 0; i < functors.size(); ++i)
{
functors[i]();
}
callingPendingFunctors_ = false;
}
// 打印所有就绪channel 事件的函数
void EventLoop::printActiveChannels() const
{
for (ChannelList::const_iterator it = activeChannels_.begin();
it != activeChannels_.end(); ++it)
{
const Channel *ch = *it;
LOG_TRACE << "{" << ch->reventsToString() << "} ";
}
}
2 什么时候需要runinloop()
eventloop所在线程是实际进行IO操作的线程。
虽然，eventloop 被绑定在每一个channel，但是实际上channel 又被关联在各种其他的类中，比如TcpConnection 中，当TcpConnection在别的线程中，需要发送数据的时候，那么会调用send，此时TcpConnection判断运行函数的线程是否是关联的channel绑定的evenloop，如果是，那么直接执行，如果不是，那么runinloop()，通过bind的方式传入。因为send需要传入数据，而在bind中绑定数据即可。
对于这点，后面再添加。
因为涉及到一个什么时候析构的问题。