简介
EventLoop.cc就相当于一个reactor，多线程之间的函数调用(用eventfd唤醒)，epoll处理，超时队列处理，对channel的处理。运行loop的进程被称为IO线程，EventLoop提供了一些API确保相应函数在IO线程中调用，确保没有用互斥量保护的变量只能在IO线程中使用，也封装了超时队列的基本操作。
成员及属性解析
一个事件循环，注意，一个创建了EventLoop对象的线程是workloop线程
主要接口
loop
死循环，阻塞在Poller的poll函数，等待唤醒唤醒后执行ChannelList中每个Channel的回调最后执行任务队列中的Functor
runInLoop
在IO线程中执行用户回调Functor，若调用者非IO线程，则会调用queueInLoop
queueInLoop
当调用者并非当前EventLoop所在线程时，将Functor存入EventLoop的任务队列从而保证Functor由IO线程执行，这是线程安全的保证之一
updateChannel与removeChannel
核心中的核心，通过这个公有接口建立起Channel和Poller沟通的桥梁Channel通过这个接口向Poller注册或者移除自己的fd实现了Poller和Channel两端的解耦
核心实现：handleEvent
遍历所有的activeChannelList_，并依次执行这些Channel中注册的回调函数这个环节非常非常关键，是一切事件派发机制中回调执行的地方
主要成员
wakeupchannel_
通过eventfd唤醒的channel
EventLoop可以通过这个Channel唤醒自己执行定时任务
activeChannelList_
通过一次poll获得的所有发生事件的Channel指针列表
pendingFunctors_
所有非IO线程调用的用户回调都会存放在这个队列中，通过mutex互斥量保护
poller_
一个多路复用实例
源码剖析
EventLoop.h
#ifndef MUDUO_NET_EVENTLOOP_H
#define MUDUO_NET_EVENTLOOP_H
#include <atomic>
#include <functional>
#include <vector>
#include <boost/any.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;
///
/// Reactor, at most one per thread.
///
/// This is an interface class, so don't expose too much details.
class EventLoop : noncopyable
{
public:
typedef std::function<void()> Functor;
EventLoop();
~EventLoop(); // force out-line dtor, for std::unique_ptr members.
//开启事件循环
void loop();
//退出事件循环
void quit();
//轮询返回的时间，通常意味着数据到达。
Timestamp pollReturnTime() const {
return pollReturnTime_; }
int64_t iteration() const {
return iteration_; }
/// Runs callback immediately in the loop thread.
/// It wakes up the loop, and run the cb.
/// If in the same loop thread, cb is run within the function.
/// Safe to call from other threads.
///在当前loop中执行cb
void runInLoop(Functor cb);
/// Queues callback in the loop thread.
/// Runs after finish pooling.
/// Safe to call from other threads.
///将cb放入队列中,唤醒loop所在的线程执行
void queueInLoop(Functor cb);
size_t queueSize() const;
// timers
///
/// Runs callback at 'time'.
/// Safe to call from other threads.
///
TimerId runAt(Timestamp time, TimerCallback cb);
///
/// Runs callback after @c delay seconds.
/// Safe to call from other threads.
///
TimerId runAfter(double delay, TimerCallback cb);
///
/// Runs callback every @c interval seconds.
/// Safe to call from other threads.
///
TimerId runEvery(double interval, TimerCallback cb);
///
/// Cancels the timer.
/// Safe to call from other threads.
///
void cancel(TimerId timerId);
// internal usage
//唤醒loop所在的线程
void wakeup();
//调用poller的方法
void updateChannel(Channel* channel);
void removeChannel(Channel* channel);
bool hasChannel(Channel* channel);
// pid_t threadId() const { return threadId_; }
void assertInLoopThread()
{
if (!isInLoopThread())
{
abortNotInLoopThread();
}
}
//判断eventloop对象是否在自己的线程
bool isInLoopThread() const {
return threadId_ == CurrentThread::tid(); }
// bool callingPendingFunctors() const { return callingPendingFunctors_; }
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();
void handleRead(); // waked up
void doPendingFunctors();//在loop一次后执行pendingFunctors_中的所有方法(会清空队列)
void printActiveChannels() const; // DEBUG
typedef std::vector<Channel*> ChannelList;
bool looping_; /* atomic */
std::atomic<bool> quit_;//标识loop的退出
bool eventHandling_; /* atomic */
//标识当前loop是否需要有执行的回调操作
bool callingPendingFunctors_; /* atomic */
int64_t iteration_;
const pid_t threadId_;//记录thread所在的线程pid
Timestamp pollReturnTime_;
std::unique_ptr<Poller> poller_;
std::unique_ptr<TimerQueue> timerQueue_;
//主要作用,当mainLoop获取到一个accept新用户的channel,通过轮询算法选择一个subloop,通过该成员唤醒subloop处理,使用eventfd
int wakeupFd_;
// unlike in TimerQueue, which is an internal class,
// we don't expose Channel to client.
std::unique_ptr<Channel> wakeupChannel_;
boost::any context_;
// scratch variables
ChannelList activeChannels_;
Channel* currentActiveChannel_;
mutable MutexLock mutex_;//保证pendingFunctors_的线程安全操作
std::vector<Functor> pendingFunctors_ GUARDED_BY(mutex_);//存储loop需要执行的所有操作
};
} // namespace net
} // namespace muduo
#endif // MUDUO_NET_EVENTLOOP_H
eventloop.cc
// Copyright 2010, Shuo Chen. All rights reserved.
// http://code.google.com/p/muduo/
//
// Use of this source code is governed by a BSD-style license
// that can be found in the License file.
// Author: Shuo Chen (chenshuo at chenshuo dot com)
#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 <algorithm>
#include <signal.h>
#include <sys/eventfd.h>
#include <unistd.h>
using namespace muduo;
using namespace muduo::net;
namespace
{
//保证一个线程只有一个loop
__thread EventLoop* t_loopInThisThread = 0;
//poll超时时间
const int kPollTimeMs = 10000;
int createEventfd()
{
int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
if (evtfd < 0)
{
LOG_SYSERR << "Failed in eventfd";
abort();
}
return evtfd;
}
#pragma GCC diagnostic ignored "-Wold-style-cast"
class IgnoreSigPipe
{
public:
IgnoreSigPipe()
{
::signal(SIGPIPE, SIG_IGN);
// LOG_TRACE << "Ignore SIGPIPE";
}
};
#pragma GCC diagnostic error "-Wold-style-cast"
IgnoreSigPipe initObj;
} // namespace
EventLoop* EventLoop::getEventLoopOfCurrentThread()
{
return t_loopInThisThread;
}
//创建了EventLoop对象的线程称为IO线程
EventLoop::EventLoop()
: looping_(false), //判断是否在loop
quit_(false), //判断是否退出的标志
eventHandling_(false), //处理handevent的标志
callingPendingFunctors_(false), //判断当前是不是在执行方法队列
iteration_(0),
threadId_(CurrentThread::tid()), //当前线程ID
poller_(Poller::newDefaultPoller(this)), //创建一个 poll 或 epoll 对象
timerQueue_(new TimerQueue(this)), //创建一个计时器
wakeupFd_(createEventfd()), //发送唤醒loop消息的描述符,随便写点消息即可唤醒
wakeupChannel_(new Channel(this, wakeupFd_)), //wakeupChannel_用来自己给自己通知的一个通道，该通道会纳入到poller来管理
currentActiveChannel_(NULL) //当前活跃的channel链表指针
{
LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;
if (t_loopInThisThread) //判断是否是本线程的loop,是一个loop类型的指针
{
LOG_FATAL << "Another EventLoop " << t_loopInThisThread
<< " exists in this thread " << threadId_; //用LOG_FATAL终止abort它
}
else
{
t_loopInThisThread = this; //this赋给线程局部数据指针
}
//设定wakeupChannel的回调函数，即EventLoop自己的的handleRead函数
wakeupChannel_->setReadCallback(
std::bind(&EventLoop::handleRead, this)); //channel->handleEventWithGuard会调用到handleRead
// we are always reading the wakeupfd
wakeupChannel_->enableReading(); //注册wakeupFd_到poller
}
EventLoop::~EventLoop()
{
LOG_DEBUG << "EventLoop " << this << " of thread " << threadId_
<< " destructs in thread " << CurrentThread::tid();
wakeupChannel_->disableAll(); //从监听队列fd里移除
wakeupChannel_->remove(); //移除epoll里面的channel
::close(wakeupFd_);
t_loopInThisThread = NULL;
}
void EventLoop::loop()
{
assert(!looping_);
assertInLoopThread(); //事件循环必须在IO线程中，即创建该evenloop的线程
looping_ = true; //是否正在循环
quit_ = false; // FIXME: what if someone calls quit() before loop() ?
LOG_TRACE << "EventLoop " << this << " start looping";
while (!quit_)
{
activeChannels_.clear(); //activeChannels_是一个vector
//等待io复用函数返回
pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_); //调用poll返回活动的事件，有可能是唤醒返回的
++iteration_;
//根据设置的日志等级打印跟踪信息
if (Logger::logLevel() <= Logger::TRACE)
{
printActiveChannels();
}
// TODO sort channel by priority 按优先级排序
//处理IO事件
eventHandling_ = true;
for (Channel* channel : activeChannels_) //遍历通道来进行处理
{
currentActiveChannel_ = channel;
currentActiveChannel_->handleEvent(pollReturnTime_); //pollReturnTime_是poll返回的时刻
}
currentActiveChannel_ = NULL; //处理完了赋空
eventHandling_ = false;
//执行方法队列中的方法[方法队列functors，我们可以跨线程的往里面添加新的方法，这些方法会在处理完io事件后执行]
doPendingFunctors(); //这个设计也能够进行计算任务
}
LOG_TRACE << "EventLoop " << this << " stop looping";
looping_ = false;
}
void EventLoop::quit()
{
quit_ = true; //设置退出标志
// There is a chance that loop() just executes while(!quit_) and exits,
// then EventLoop destructs, then we are accessing an invalid object.
// Can be fixed using mutex_ in both places.
if (!isInLoopThread())
{
wakeup(); //唤醒
}
}
//在I/O线程中调用某个函数
//实际上就是如果是I/O线程主动调用该函数想要执行，那就同步执行该函数。如果是其他线程施加给I/O线程的任务，那么其他线程就需要把回调函数加入I/O线程的队列，等待异步执行
void EventLoop::runInLoop(Functor cb)
{
if (isInLoopThread()) //判断是否是本线程的loop
{
cb();
}
else
{
queueInLoop(std::move(cb));
}
}
void EventLoop::queueInLoop(Functor cb)//把方法添加到队列中，该方法会出现在多个线程中,操作要加锁
{
{
MutexLockGuard lock(mutex_);
pendingFunctors_.push_back(std::move(cb));//std::function支持移动初始化，所以这里用move提升性能。(减少一次拷贝)
}
if (!isInLoopThread() || callingPendingFunctors_)//如果调用的queneInLoop的线程不是IO线程，那么唤醒
{
//如果在IO线程调用queueInLoop()，而此时正在调用pending functor,由于doPendingFunctors()调用的Functor可能再次调用queueInLoop(cb)，这是queueInLoop()就必须wakeup(),否则新增的cb可能就不能及时调用了
wakeup();
}
}
size_t EventLoop::queueSize() const
{
MutexLockGuard lock(mutex_);
return pendingFunctors_.size();
}
TimerId EventLoop::runAt(Timestamp time, TimerCallback cb)//在指定的时间调用callback
{
return timerQueue_->addTimer(std::move(cb), time, 0.0);
}
TimerId EventLoop::runAfter(double delay, TimerCallback cb)//等一段时间调用callback
{
Timestamp time(addTime(Timestamp::now(), delay));//微妙
return runAt(time, std::move(cb));
}
TimerId EventLoop::runEvery(double interval, TimerCallback cb)//以固定的间隔反复的调用callback
{
Timestamp time(addTime(Timestamp::now(), interval));
return timerQueue_->addTimer(std::move(cb), time, interval);
}
void EventLoop::cancel(TimerId timerId) //取消timer
{
return timerQueue_->cancel(timerId);
}
void EventLoop::updateChannel(Channel* channel) //更新通道,用epoll_ctl更新fd
{
assert(channel->ownerLoop() == this); //判断channel的loop是不是当前loop
assertInLoopThread();
poller_->updateChannel(channel);
}
void EventLoop::removeChannel(Channel* channel) //移除通道,将channel从ChannelMap移除并EPOLL_CTL_DEL掉fd
{
assert(channel->ownerLoop() == this); //表示当前的loop
assertInLoopThread();
if (eventHandling_) //正在处理channel
{
assert(currentActiveChannel_ == channel || //当前的channel或不是活跃的channel
std::find(activeChannels_.begin(), activeChannels_.end(), channel) == activeChannels_.end());
}
poller_->removeChannel(channel);
}
bool EventLoop::hasChannel(Channel* channel)//查找事件分发器是否在channels_中
{
assert(channel->ownerLoop() == this);
assertInLoopThread();
return poller_->hasChannel(channel);
}
void EventLoop::abortNotInLoopThread()
{
LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
<< " was created in threadId_ = " << threadId_
<< ", current thread id = " << CurrentThread::tid();
}
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";
}
}
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";
}
}
// 1. 不是简单的在临界区内依次调用functor，而是把回调列表swap到functors中，这一方面减小了
//临界区的长度，意味着不会阻塞其他线程的queueInLoop()，另一方面也避免了死锁(因为Functor可能再次调用quueInLoop)
// 2. 由于doPendingFunctors()调用的Functor可能再次调用queueInLoop(cb)，这是queueInLoop()就必须wakeup(),否则新增的cb可能就不能及时调用了
// 3. muduo没有反复执行doPendingFunctors()直到pendingFunctors为空，这是有意的，否则I/O线程可能陷入死循环，无法处理I/O事件
void EventLoop::doPendingFunctors()
{
std::vector<Functor> functors;
callingPendingFunctors_ = true;
//注意这里的临界区，这里使用了一个栈上变量functors和pendingFunctors交换
{
MutexLockGuard lock(mutex_);
functors.swap(pendingFunctors_); //pendingFunctors_是存放Functor的vector
}
//此处其它线程就可以往pendingFunctors添加任务
for (const Functor& functor : functors)
{
functor();
}
callingPendingFunctors_ = false;
}
void EventLoop::printActiveChannels() const
{
for (const Channel* channel : activeChannels_)
{
LOG_TRACE << "{" << channel->reventsToString() << "} ";
}
}
