C++的学习过程充满着迷茫，C++primer，侯捷老师的STL源码剖析，再到boost库，C++多线程库，纷繁复杂的数据结构，以及为了效率无所不用其极的函数重载，shared_ptr的线程安全性，多线程访问时数据竞争，C++对于C的进步在于安全的内存管理，以及大量库支持，避免了我们重复造轮子的过程。最近看了陈硕老师的muduo网络库，该网络库是一个C++网络服务中的经典，muduo库的意义在于：“奋木铎以警众，使明听也”，为了不至于看久了忘记，就撰写这篇博文，本博客将会从程硕老师的出版图书《Linux多线程服务端编程 使用muduo C++网络库》的第八章开始讲解（某东上打折很给力，买来看看还是很值得的！希望大家支持正版）。
在正式介绍之前不妨先思考一下网络库的设计需求，设立几个小目标。
1.一个单线程的事件处理机制。
2.一个基于网络通信的事件处理（举例来说，对通信时间进行处理以及处理后发送的处理机制）。
3.多线程下实现上述1，2要求。
接下来将分为三个章节来讲述要求1,2,3，的源码解读。
muduo网络库将网络服务通信的各个流程中的环节封装成不同的class，高度模块化的设计使得以后拓展接口提供了极大地便利，class EventLoop是网络库的重要组成部分，初始EventLoop什么也不做，代码如下：
#ifndef MUDUO_NET_EVENTLOOP_H
#define MUDUO_NET_EVENTLOOP_H
#include "thread/Thread.h"
namespace muduo
{
class EventLoop : boost::noncopyable
{
public:
EventLoop();
~EventLoop();
void loop();
void assertInLoopThread()
{
if (!isInLoopThread())
{
abortNotInLoopThread();
}
}
bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }
private:
void abortNotInLoopThread();
bool looping_;
const pid_t threadId_;
};
}
#endif
可以看到EventLoop定义有构造，析构，loop，isInLoopThread，abortNotInLoopThread等函数，成员函数的定义如下：
#include "EventLoop.h"
#include "logging/Logging.h"
#include <assert.h>
#include <poll.h>
using namespace muduo;
__thread EventLoop* t_loopInThisThread = 0;//thread_local变量
EventLoop::EventLoop()
: looping_(false),
threadId_(CurrentThread::tid())
{
LOG_TRACE << "EventLoop created " << this << " in thread " << threadId_;
if (t_loopInThisThread)
{
LOG_FATAL << "Another EventLoop " << t_loopInThisThread
<< " exists in this thread " << threadId_;
}
else
{
t_loopInThisThread = this;
}
}
EventLoop::~EventLoop()
{
assert(!looping_);
t_loopInThisThread = NULL;
}
void EventLoop::loop()
{
assert(!looping_);
assertInLoopThread();
looping_ = true;
::poll(NULL, 0, 5*1000);
LOG_TRACE << "EventLoop " << this << " stop looping";
looping_ = false;
}
void EventLoop::abortNotInLoopThread()
{
LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
<< " was created in threadId_ = " << threadId_
<< ", current thread id = " << CurrentThread::tid();
}
其中t_loopInThisThread属于thread变量，存储当前线程的EventLoop指针，初始为0，在构造函数中，对事件循环的looping_标志位，以及线程号进行初始化（一个EventLoop对应一个线程，避免多线程下的数据竞争以及访问逻辑混乱的情况，所以使用大量assertInLoopThread以及boost：noncopyable以保证EventLoop在本线程中运行），LOG_TRACE 等属于日志，便于意外情况下对服务器运行情况的分析，以后不再赘述。
析构函数太过简单不需要解释；loop（）函数用于正式启用事件循环，本例中loop（）并不完全，poll函数属于IO复用，之后会讲解。abortNotInLoopThread()将线程运行错误输入日志。以上便是一个EventLoop的基本结构，但并不涉及具体的事件处理机制。
在介绍class Channel之前，不得不说说Linux下IO复用——poll函数。
int poll(struct pollfd *fds, nfds_t nfds, int timeout);
pollfd为poll函数需要监听的事件，nfds为需要监听事件的数量，pollfd为
struct pollfd{
int fd; //文件描述符
short events; //需要监听的事件
short revents; //实际发生的事件
};
Linux一大特色便是万物皆文件，int fd指定了一个文件描述符，poll函数会根据events（用户设置）来监听fd，并根据发生的事件重写revents，并返回具体事件发生的数量。
因此对于上述调用poll环节可以分为class Channel和class Poller两部分！以下为class Channel的源码：
#ifndef MUDUO_NET_CHANNEL_H
#define MUDUO_NET_CHANNEL_H
#include <boost/function.hpp>
#include <boost/noncopyable.hpp>
namespace muduo
{
class EventLoop;
class Channel : boost::noncopyable
{
public:
typedef boost::function<void()> EventCallback;//定义回调函数
Channel(EventLoop* loop, int fd);
void handleEvent();//事件具体处理
void setReadCallback(const EventCallback& cb)//读回调函数设置
{ readCallback_ = cb; }
void setWriteCallback(const EventCallback& cb)//写回调函数设置
{ writeCallback_ = cb; }
void setErrorCallback(const EventCallback& cb)//错误回调函数设置
{ errorCallback_ = cb; }
int fd() const { return fd_; }
int events() const { return events_; }
void set_revents(int revt) { revents_ = revt; }
bool isNoneEvent() const { return events_ == kNoneEvent; }
void enableReading() { events_ |= kReadEvent; update(); }//该函数负责向poller的channal列表进行注册
// void enableWriting() { events_ |= kWriteEvent; update(); }
// void disableWriting() { events_ &= ~kWriteEvent; update(); }
// void disableAll() { events_ = kNoneEvent; update(); }
// for Poller
int index() { return index_; }
void set_index(int idx) { index_ = idx; }
EventLoop* ownerLoop() { return loop_; }
private:
void update();
static const int kNoneEvent;//
static const int kReadEvent;//
static const int kWriteEvent;//
EventLoop* loop_;//所属的EventLoop
const int fd_;
int events_;
int revents_;
int index_; // used by Poller.
EventCallback readCallback_;//函数指针
EventCallback writeCallback_;//函数指针
EventCallback errorCallback_;//函数指针
};
成员函数的实现：
#include "Channel.h"
#include "EventLoop.h"
#include "logging/Logging.h"
#include <sstream>
#include <poll.h>
using namespace muduo;
const int Channel::kNoneEvent = 0;
const int Channel::kReadEvent = POLLIN | POLLPRI;
const int Channel::kWriteEvent = POLLOUT;
Channel::Channel(EventLoop* loop, int fdArg)
: loop_(loop),
fd_(fdArg),
events_(0),
revents_(0),
index_(-1)
{
}
void Channel::update()
{
loop_->updateChannel(this);
}
void Channel::handleEvent()//根据poll返回的事件revents_标识符调用相应的事件回调函数
{
if (revents_ & POLLNVAL) {
LOG_WARN << "Channel::handle_event() POLLNVAL";
}
if (revents_ & (POLLERR | POLLNVAL)) {
if (errorCallback_) errorCallback_();
}
if (revents_ & (POLLIN | POLLPRI | POLLRDHUP)) {
if (readCallback_) readCallback_();
}
if (revents_ & POLLOUT) {
if (writeCallback_) writeCallback_();
}
}
可以看出，class Channel的功能主要在于创建事件pollfd以及相应的回调函数（一个需要监听的事件pollfd对应一个Channel，一个Channel也唯一属于一个EventLoop，相同的事件不需要在不同的EventLoop中重复的监听，逻辑上是没必要的，因此一个事件的Channel具有唯一性，boost::noncopyable），唯一没有介绍的是 enableReading()以及int index，enableReading()代表一个Channel实例已经可读，并调用自身的私有函数update（），继而调用EventLoop的updateChannel()成员函数向EventLoop注册自己，表示自己需要加入到poll的监听队列（此处的队列是一种说法，并非数据结构）中，index用于表明在监听队列中的编号。
那么问题来了，具体内部注册具有是怎么实现的呢？不妨将这个问题设为问题1
根据上述问题，EventLoop则需要改变一下，增加新的成员函数，新的代码如下：
#ifndef MUDUO_NET_EVENTLOOP_H
#define MUDUO_NET_EVENTLOOP_H
#include "thread/Thread.h"
#include <boost/scoped_ptr.hpp>
#include <vector>
namespace muduo
{
class Channel;
class Poller;
class EventLoop : boost::noncopyable
{
public:
EventLoop();
// force out-line dtor, for scoped_ptr members.
~EventLoop();
///
/// Loops forever.
///
/// Must be called in the same thread as creation of the object.
///
void loop();
void quit();//新增
// internal use only
void updateChannel(Channel* channel);//新增
// void removeChannel(Channel* channel);
void assertInLoopThread()
{
if (!isInLoopThread())
{
abortNotInLoopThread();
}
}
bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }
private:
void abortNotInLoopThread();
typedef std::vector<Channel*> ChannelList;//定义保存数据的结构
bool looping_; //循环标志位，涉及loop（）中的逻辑，使得loop（）无法重复运行。
bool quit_; //停止标志位
const pid_t threadId_;
boost::scoped_ptr<Poller> poller_;//拥有一个class Poller的实例
ChannelList activeChannels_;//活动事件的列表
};
}
#endif
成员函数如下：
#include "EventLoop.h"
#include "Channel.h"
#include "Poller.h"
#include "logging/Logging.h"
#include <assert.h>
using namespace muduo;
__thread EventLoop* t_loopInThisThread = 0;
const int kPollTimeMs = 10000;
EventLoop::EventLoop()
: looping_(false),
quit_(false),
threadId_(CurrentThread::tid()),
poller_(new Poller(this))
{
LOG_TRACE << "EventLoop created " << this << " in thread " << threadId_;
if (t_loopInThisThread)
{
LOG_FATAL << "Another EventLoop " << t_loopInThisThread
<< " exists in this thread " << threadId_;
}
else
{
t_loopInThisThread = this;
}
}
EventLoop::~EventLoop()
{
assert(!looping_);
t_loopInThisThread = NULL;
}
void EventLoop::loop()//loop（）函数正式增加了poll监听事件的结构
{
assert(!looping_);
assertInLoopThread();
looping_ = true;
quit_ = false;
while (!quit_)
{
activeChannels_.clear();//上一次的活跃Channel删除；
poller_->poll(kPollTimeMs, &activeChannels_);//调用Class Poller的成员函数，返回活跃的Channel数组
for (ChannelList::iterator it = activeChannels_.begin();
it != activeChannels_.end(); ++it)
{
(*it)->handleEvent();//调用Channel对应的回调函数
}
}
LOG_TRACE << "EventLoop " << this << " stop looping";
looping_ = false;
}
void EventLoop::quit()
{
quit_ = true;
// wakeup();
}
void EventLoop::updateChannel(Channel* channel)
{
assert(channel->ownerLoop() == this);
assertInLoopThread();
poller_->updateChannel(channel);//注册一个Channel的工作转移到了class Poller
}
void EventLoop::abortNotInLoopThread()
{
LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
<< " was created in threadId_ = " << threadId_
<< ", current thread id = " << CurrentThread::tid();
}
以上的代码并没有增加太多的成员函数，但也没有解释问题1——Channel是怎么注册的（EventLoop把问题推给了Class Poller），同时在loop（）函数中调用了Class Poller的成员函数，私有成员中增加boost::scoped_ptr poller_（这很好理解，Poller内部封装由poll的功能，EventLoop独自拥有一个Class Poller的实例就可以实现监听所有的Channel），至此，所有的问题都推给了Class Poller，代码如下：
#ifndef MUDUO_NET_POLLER_H
#define MUDUO_NET_POLLER_H
#include <map>
#include <vector>
#include "datetime/Timestamp.h"
#include "EventLoop.h"
struct pollfd;
namespace muduo
{
class Channel;
class Poller : boost::noncopyable
{
public:
typedef std::vector<Channel*> ChannelList;
Poller(EventLoop* loop);
~Poller();
Timestamp poll(int timeoutMs, ChannelList* activeChannels);//poll成员函数，与上文的poll函数不一样。
void updateChannel(Channel* channel);//向Class Poller注册Channel；
void assertInLoopThread() { ownerLoop_->assertInLoopThread(); }//emmm，不必多说
private:
void fillActiveChannels(int numEvents,
ChannelList* activeChannels) const;//填充活跃的Channel
typedef std::vector<struct pollfd> PollFdList;//定义了存储pollfd的结构，注意为vector。
typedef std::map<int, Channel*> ChannelMap;//Channel的存储结构，Channel并不直接存储pollfd，而是存储struct pollfd中对应的成员，map<int, Channel*>中int为pollfd->fd,也就是时class Channel的成员fd_。
EventLoop* ownerLoop_;//
PollFdList pollfds_;//
ChannelMap channels_;//
};
}
#endif
上述中pollfds_为vector，用于高效地随机访问（Channel中保存了在vector中的下标index），channels_使用map来管理，实现高效地查找，删除和插入Channel（尽管当前没有实现删除）。
再看看成员函数的实现：
#include "Poller.h"
#include "Channel.h"
#include "logging/Logging.h"
#include <assert.h>
#include <poll.h>
using namespace muduo;
Poller::Poller(EventLoop* loop)
: ownerLoop_(loop)//保存有所属的EventLoop指针
{
}
Poller::~Poller()//析构函数什么也不做，不要惊讶，其成员都是stl中的容器，容器内保存的是一个类的实例则自动调用其析构，具体可参考《stl源码剖析》的2.2.3小节；
//如果为指针，需要手动管理析构，但Channel的创建和析构都不属于class Poller的工作范畴，具体可参考下文class TimerQueue的析构。
{
}
Timestamp Poller::poll(int timeoutMs, ChannelList* activeChannels)
{
int numEvents = ::poll(&*pollfds_.begin(), pollfds_.size(), timeoutMs);//监听pollfds_的文件描述符，并返回发生事件
Timestamp now(Timestamp::now());//时间戳，之后会介绍
if (numEvents > 0)//有事件发生
{
LOG_TRACE << numEvents << " events happended";
fillActiveChannels(numEvents, activeChannels);//调用该函数，将活跃的Channel填充至activeChannels中
} else if (numEvents == 0) {
LOG_TRACE << " nothing happended";
} else {
LOG_SYSERR << "Poller::poll()";
}
return now;//返回当前时间戳
}
void Poller::fillActiveChannels(int numEvents,
ChannelList* activeChannels) const//填充活跃的pollfd对应的Channel添加至activeChannels中
{
for (PollFdList::const_iterator pfd = pollfds_.begin();
pfd != pollfds_.end() && numEvents > 0; ++pfd)//历遍寻找活跃的pollfd
{
if (pfd->revents > 0)
{
--numEvents;//numEvents活跃事件数量全部找到则可提前结束历遍。
ChannelMap::const_iterator ch = channels_.find(pfd->fd);//
assert(ch != channels_.end());
Channel* channel = ch->second;
assert(channel->fd() == pfd->fd);
channel->set_revents(pfd->revents);//填充fd的revents
// pfd->revents = 0;
activeChannels->push_back(channel);//填充活跃的Channel至activeChannels数组中。
}
}
}
void Poller::updateChannel(Channel* channel)//添加或删除Channel
{
assertInLoopThread();
LOG_TRACE << "fd = " << channel->fd() << " events = " << channel->events();
if (channel->index() < 0)//Channel的index默认为-1，代表未使用过
{
// a new one, add to pollfds_
assert(channels_.find(channel->fd()) == channels_.end());
struct pollfd pfd;//定义结构体pollfd，将Channel的fd，events赋予pfd中
pfd.fd = channel->fd();
pfd.events = static_cast<short>(channel->events());
pfd.revents = 0;
pollfds_.push_back(pfd);//pfd将入需要监听的pollfds_数组中。
int idx = static_cast<int>(pollfds_.size())-1;//根据在pollfds_数组的位置，设置Channel的index，此时index由-1变为pollfds_.size()-1，代表已经添加过。
channel->set_index(idx);
channels_[pfd.fd] = channel;//将入map中，可以认为是map<fd，Channel*>
} else {
// 该Channel已经被添加过一次，则实现对应的Channel更新
assert(channels_.find(channel->fd()) != channels_.end());
assert(channels_[channel->fd()] == channel);
int idx = channel->index();//得到Channel的index，即在pollfds_中的位置
assert(0 <= idx && idx < static_cast<int>(pollfds_.size()));
struct pollfd& pfd = pollfds_[idx];
assert(pfd.fd == channel->fd() || pfd.fd == -1);
pfd.events = static_cast<short>(channel->events());//更新events
pfd.revents = 0;
if (channel->isNoneEvent()) {
// ignore this pollfd
pfd.fd = -1;
}
}
}
class Poller的主要作用是维护EventLoop所拥有的的Channel，其成员函数poll用于事件的监听，并返回活跃事件，以及updateChannel（）函数向Poller中添加Channel，不过当前的class Poller不具有移除Channel的作用，只能不断的向pollfds_和channels_添加。
以下为三个class的关键函数loop（）和enableReading（）的调用循序：
为了给EventLoop根据一定周期执行某个函数，设计出响应的定时器功能，定时器由class TimerId，Timer，TimerQueue，三个来实现的，接下来将一一介绍。
首先需要明确两点：1、定时器的作用，无论EventLoop由多少个定时任务，仅需要一个class就可以集中的管理；2、poll是阻塞的，一旦监听的文件描述符没有发生事件，则会阻塞，那么定时任务无法得到及时的执行，所以需要将计时任务作为一个监听事件（也就是Channel）来实现定时的唤醒，计算有哪些计时器到期，执行对应的函数。这样我们就明确了class TimerQueue的功能了！
但在介绍class TimerQueue之前，需要先介绍几个class以便更好地理解定时器的实现：
class Timestamp的主要作用是保存时间戳（int64_t microSecondsSinceEpoch_;），以及重载了一些运算符，提供了一些如toString()，now（）等接口，具体的成员函数的实现不讲了，简单讲一下头文件中成员的作用。
#ifndef MUDUO_BASE_TIMESTAMP_H
#define MUDUO_BASE_TIMESTAMP_H
#include "copyable.h"
#include <stdint.h>
#include <string>
namespace muduo
{
class Timestamp : public muduo::copyable
{
public:
Timestamp();//默认构造函数，时间戳设为0
explicit Timestamp(int64_t microSecondsSinceEpoch);//初始化时间戳为microSecondsSinceEpoch
void swap(Timestamp& that)//交换时间戳
{
std::swap(microSecondsSinceEpoch_, that.microSecondsSinceEpoch_);
}
std::string toString() const;//时间戳转为str
std::string toFormattedString() const;//同上类似
bool valid() const { return microSecondsSinceEpoch_ > 0; }//时间戳是否可用
int64_t microSecondsSinceEpoch() const { return microSecondsSinceEpoch_; }//返回时间戳
static Timestamp now();//静态函数，返回当前时间的时间戳
static Timestamp invalid();//返回一个为0的时间戳
static const int kMicroSecondsPerSecond = 1000 * 1000;
private:
int64_t microSecondsSinceEpoch_;//时间戳以微妙来存储
};
inline bool operator<(Timestamp lhs, Timestamp rhs)//重载运算符
{
return lhs.microSecondsSinceEpoch() < rhs.microSecondsSinceEpoch();
}
inline bool operator==(Timestamp lhs, Timestamp rhs)
{
return lhs.microSecondsSinceEpoch() == rhs.microSecondsSinceEpoch();
}
inline double timeDifference(Timestamp high, Timestamp low)//时间差
{
int64_t diff = high.microSecondsSinceEpoch() - low.microSecondsSinceEpoch();
return static_cast<double>(diff) / Timestamp::kMicroSecondsPerSecond;
}
inline Timestamp addTime(Timestamp timestamp, double seconds)//为给定的时间戳增加seconds秒，并返回增加后的时间戳
{
int64_t delta = static_cast<int64_t>(seconds * Timestamp::kMicroSecondsPerSecond);
return Timestamp(timestamp.microSecondsSinceEpoch() + delta);
}
}
#endif
上小节class Timestamp构成class Timer的成员，class Timer顾名思义：定时器！首先来明确定时器的作用，EventLoop可以拥有数个定时任务，那么单个定时器不仅需要保存时间戳，定时时间，是否循环，以及相应的时间回调函数。一下为class Timer的头文件：
#ifndef MUDUO_NET_TIMER_H
#define MUDUO_NET_TIMER_H
#include <boost/noncopyable.hpp>
#include "datetime/Timestamp.h"
#include "Callbacks.h"
namespace muduo
{
class Timer : boost::noncopyable//不可拷贝
{
public:
Timer(const TimerCallback& cb, Timestamp when, double interval)//构造函数，分别设定回调函数，时间戳，循环间隔，是否重复
: callback_(cb),
expiration_(when),
interval_(interval),
repeat_(interval > 0.0)
{ }
void run() const//运行回调函数
{
callback_();
}
Timestamp expiration() const { return expiration_; }//返回时间戳
bool repeat() const { return repeat_; }//返回是否重复
void restart(Timestamp now);//重启
{
if (repeat_)
{
expiration_ = addTime(now, interval_);//默认=调用
}
else
{
expiration_ = Timestamp::invalid();//默认=调用
}
}
private:
const TimerCallback callback_;
Timestamp expiration_;
const double interval_;
const bool repeat_;
};
}
#endif
接下来就是class TimerQueue的正体了，class TimerQueue为定时器队列，先来明确class TimerQueue的实现任务：1、提供对外接口——增加定时任务；2、创建一个timerfd（LInux新增了timerfd作为定时任务，用法和上文的pollfd一样，当超时后发生可读事件，所以同样需要Channel，暂且把这个Channel叫他timerqueue Channel，timerqueue Channel对应一个回调函数），并使用poll来监听，对该文件描述符可读事件进行超时通知，从poll返回后调用timerqueue Channel的回调函数，该回调函数会检查class TimerQueue所有定时器是否到期，调用其到期的Timer的回调函数。上代码直接看：
#ifndef MUDUO_NET_TIMERQUEUE_H
#define MUDUO_NET_TIMERQUEUE_H
#include <set>
#include <vector>
#include <boost/noncopyable.hpp>
#include "datetime/Timestamp.h"
#include "thread/Mutex.h"
#include "Callbacks.h"
#include "Channel.h"
namespace muduo
{
class EventLoop;
class Timer;
class TimerId;
class TimerQueue : boost::noncopyable//一个EventLoop对应一个定时器队列
{
public:
TimerQueue(EventLoop* loop);
~TimerQueue();
TimerId addTimer(const TimerCallback& cb,
Timestamp when,
double interval);//对外接口，增加定时器，以及设定相关的Timer的回调函数和定时间隔。
private:
typedef std::pair<Timestamp, Timer*> Entry;
typedef std::set<Entry> TimerList;//set内置红黑树，且自动排序，方便找到已经到期的Timer
void handleRead();//TimerQueue的回调函数，poll返回后运行，检查其到期的定时器，并运行Timer的回调函数
std::vector<Entry> getExpired(Timestamp now);//以数组形式返回到期的Timer，
void reset(const std::vector<Entry>& expired, Timestamp now);//重新设置定时器Timer的时间戳，
bool insert(Timer* timer);//插入定时器Timer
EventLoop* loop_;//所属的EventLoop
const int timerfd_;//timefd的文件描述符
Channel timerfdChannel_;//timefd对应的Channel实例
TimerList timers_;//保存定时器Timer
};
}
#endif // MUDUO_NET_TIMERQUEUE_H
成员函数定义如下：
#define __STDC_LIMIT_MACROS
#include "TimerQueue.h"
#include "logging/Logging.h"
#include "EventLoop.h"
#include "Timer.h"
#include "TimerId.h"
#include <boost/bind.hpp>
#include <sys/timerfd.h>
namespace muduo
{
namespace detail
{
int createTimerfd()//非成员函数，创建timerfd文件描述符用于poll监听，注意没有设置超时时间
{
int timerfd = ::timerfd_create(CLOCK_MONOTONIC,
TFD_NONBLOCK | TFD_CLOEXEC);
if (timerfd < 0)
{
LOG_SYSFATAL << "Failed in timerfd_create";
}
return timerfd;//返回timefd的文件描述符
}
struct timespec howMuchTimeFromNow(Timestamp when)//void resetTimerfd（...）需要的函数，用于计算timerfd的超时时间
{
int64_t microseconds = when.microSecondsSinceEpoch()
- Timestamp::now().microSecondsSinceEpoch();
if (microseconds < 100)
{
microseconds = 100;
}
struct timespec ts;
ts.tv_sec = static_cast<time_t>(
microseconds / Timestamp::kMicroSecondsPerSecond);
ts.tv_nsec = static_cast<long>(
(microseconds % Timestamp::kMicroSecondsPerSecond) * 1000);
return ts;
}
void readTimerfd(int timerfd, Timestamp now)//
{
uint64_t howmany;
ssize_t n = ::read(timerfd, &howmany, sizeof howmany);
LOG_TRACE << "TimerQueue::handleRead() " << howmany << " at " << now.toString();
if (n != sizeof howmany)
{
LOG_ERROR << "TimerQueue::handleRead() reads " << n << " bytes instead of 8";
}
}
void resetTimerfd(int timerfd, Timestamp expiration)//设置timerfd的超时时间，每次poll返回timerfd后都需要重新设置，以保持及时唤醒定时器！
{
// wake up loop by timerfd_settime()
struct itimerspec newValue;
struct itimerspec oldValue;
bzero(&newValue, sizeof newValue);
bzero(&oldValue, sizeof oldValue);
newValue.it_value = howMuchTimeFromNow(expiration);
int ret = ::timerfd_settime(timerfd, 0, &newValue, &oldValue);
if (ret)
{
LOG_SYSERR << "timerfd_settime()";
}
}
}
}
using namespace muduo;
using namespace muduo::detail;
TimerQueue::TimerQueue(EventLoop* loop)
: loop_(loop),//设置所属的EventLoop
timerfd_(createTimerfd()),//创建一个，也是唯一一个timerfd，定时器文件描述符，用于poll的IO复用
timerfdChannel_(loop, timerfd_),//创建一个Channel，Channel是联系class Poller唯一途径。
timers_()//定时器class timer的队列
{
timerfdChannel_.setReadCallback(
boost::bind(&TimerQueue::handleRead, this));//设置Channel的可读事件的回调函数
timerfdChannel_.enableReading();//向EventLoop的成员Poller进行注册timerfd
}
TimerQueue::~TimerQueue()
{
::close(timerfd_);//关闭tiemrfd，该资源由class TimerQueue创建，声明周期与其一致
for (TimerList::iterator it = timers_.begin();
it != timers_.end(); ++it)
{
delete it->second;//class Timer的实例由class TimerQueue创建，需要回收其资源
}
}
TimerId TimerQueue::addTimer(const TimerCallback& cb,
Timestamp when,
double interval)//添加class timer的实例
{
Timer* timer = new Timer(cb, when, interval);//需要负责其析构
loop_->assertInLoopThread();
bool earliestChanged = insert(timer);//插入set<entry>中，并返回一个bool值，表示是否需要重新设置timerfd的超时时间
if (earliestChanged)
{
resetTimerfd(timerfd_, timer->expiration());
}
return TimerId(timer);
}
void TimerQueue::handleRead()//handleRead()为向Channel注册的可读事件的回调函数，主要功能是找到到期的Timer，并执行响应的Timer中的回调函数
{
loop_->assertInLoopThread();
Timestamp now(Timestamp::now());
readTimerfd(timerfd_, now);
std::vector<Entry> expired = getExpired(now);//getExpired（）返回到期的Timer的vector
// safe to callback outside critical section
for (std::vector<Entry>::iterator it = expired.begin();
it != expired.end(); ++it)
{
it->second->run();//执行Timer的回调函数
}
reset(expired, now);//重新将Timer加入到set<Entry>中，set会自动对定时器的时间卓先后顺序排序
}
std::vector<TimerQueue::Entry> TimerQueue::getExpired(Timestamp now)//返回到期的Timer的vector
{
std::vector<Entry> expired;
Entry sentry = std::make_pair(now, reinterpret_cast<Timer*>(UINTPTR_MAX));//设置到期的最大值
TimerList::iterator it = timers_.lower_bound(sentry);//找到lower_bound
assert(it == timers_.end() || now < it->first);
std::copy(timers_.begin(), it, back_inserter(expired));//将到期Timer拷贝至vector<Entry> expired中
timers_.erase(timers_.begin(), it);删除set<Entry>中到期Timer
return expired;//返回到期的Timer
}
void TimerQueue::reset(const std::vector<Entry>& expired, Timestamp now)//将expired中的Timer根据循环间隔重新加入到set<Entry>的Timer队列
{
Timestamp nextExpire;
for (std::vector<Entry>::const_iterator it = expired.begin();
it != expired.end(); ++it)
{
if (it->second->repeat())
{
it->second->restart(now);
insert(it->second);
}
else
{
// FIXME move to a free list
delete it->second;
}
}
if (!timers_.empty())
{
nextExpire = timers_.begin()->second->expiration();
}
if (nextExpire.valid())
{
resetTimerfd(timerfd_, nextExpire);
}
}
bool TimerQueue::insert(Timer* timer)//插入一个timer
{
bool earliestChanged = false;
Timestamp when = timer->expiration();//返回时间戳
TimerList::iterator it = timers_.begin();
if (it == timers_.end() || when < it->first)//set<Entry>timers_的第一Entry为最小的时间戳
{
earliestChanged = true;//需要重设timerfd的超时时间的bool标志
}
std::pair<TimerList::iterator, bool> result =
timers_.insert(std::make_pair(when, timer));//将Timer插入到set<Entry>timers_中
assert(result.second);
return earliestChanged;//返回需要重设timerfd的超时时间的bool标志
}
洋洋洒洒标注了那么多，估计会看晕，但要点只有3个：
1、从构造函数上看，只有创建timerfd，创建Channel，注册回调函数和Channel，之后的事情就是其他class的问题了。
2、而另一个重要的功能就是添加定时器——TimerQueue::addTimer(const TimerCallback& cb,…)，该函数添加一个定时器需要维护timerfd的超时时间（超时时间需要根据timers中最小的时间戳来决定），以及一个定时器列表，该列表根据时间戳从小到大排列，这也是为什么该列表使用set来实现的且set<pair<timestamp,timer*>>中pair<timestamp,timer*>不会出现重复。
3、timerfd超时后调用的回调函数handleRead()，该函数会从set<pair<timestamp,timer*>>取出到期的timer执行相应的回调函数timecallback（），执行完毕后计算timer下次到期的时间戳，再次加入到set<pair<timestamp,timer*>>中。
实际流程图如下：
构造函数注册
poll对timerfd超时后的调用顺序
addtimer函数的调用顺序就不写了，基本也就那么个顺序。
TimerQueue对外接口的函数只有一个，addtimer（），只要在EventLoop中加入TimerQueue，定义相关调用addtimer（）成员函数就可实现定时任务。
EventLoop新增：
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);
}
private:
boost::scoped_ptr<TimerQueue> timerQueue_;
至此定时器暂且还算是大功告成， 需要注意的是，为了不造成数据竞争等严重问题，addtimer（）加入了loop_->assertInLoopThread()，因为如果有两个线程A，B，A线程调用addtimer（），B线程为EventLoop的所属线程，并且B线程正在调用了reset（）或getExpired(Timestamp now)，以上三个函数都会对set容器进行操作，线程安全不法保证！ 但加入了断言，使得A线程无法随心所欲的使用addtimer等函数，解决方法是EventLoop加入一个函数队列，如果不在函数执行不在B线程中则加入B线程中EventLoop的函数队列，B线程在poll阻塞结束后执行完时间，再执行函数队列中的函数，当然得考虑B线程一直阻塞的情况，那么这个函数队列永远得不到执行，继而需要增加一个Channel，用于将poll从阻塞中及时唤醒。
那么新的EventLoop如下：
class EventLoop : boost::noncopyable
{
public:
typedef boost::function<void()> Functor;
EventLoop();
~EventLoop();
void loop();
void quit();
Timestamp pollReturnTime() const { return pollReturnTime_; }
void runInLoop(const Functor& cb);//可以暴露给其他线程的成员，线程安全性得以保证。
void queueInLoop(const Functor& cb);//将函数加入函数队列
TimerId runAt(const Timestamp& time, const TimerCallback& cb);
TimerId runAfter(double delay, const TimerCallback& cb);
TimerId runEvery(double interval, const TimerCallback& cb);
void wakeup();//使线程从poll阻塞中返回
void updateChannel(Channel* channel);
void assertInLoopThread()
{
if (!isInLoopThread())
{
abortNotInLoopThread();
}
}
bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }
private:
void abortNotInLoopThread();
void handleRead(); // 事件回调
void doPendingFunctors();//执行函数队列
typedef std::vector<Channel*> ChannelList;
bool looping_; /* atomic */
bool quit_; /* atomic */
bool callingPendingFunctors_; //正在执行函数队列的标识符，具有深意，能够及时的wakeup（）
const pid_t threadId_;
Timestamp pollReturnTime_;
boost::scoped_ptr<Poller> poller_;
boost::scoped_ptr<TimerQueue> timerQueue_;
int wakeupFd_;//文件描述符，属于EventLoop，对wakeupFd_写入后，poll会从阻塞中返回，及时的唤醒
boost::scoped_ptr<Channel> wakeupChannel_;
ChannelList activeChannels_;
MutexLock mutex_;//锁
std::vector<Functor> pendingFunctors_; // 函数队列
};
}
#endif
成员函数如下：
#include "EventLoop.h"
#include "Channel.h"
#include "Poller.h"
#include "TimerQueue.h"
#include "logging/Logging.h"
#include <boost/bind.hpp>
#include <assert.h>
#include <sys/eventfd.h>
using namespace muduo;
__thread EventLoop* t_loopInThisThread = 0;
const int kPollTimeMs = 10000;
static int createEventfd()//创建一个文件描述符，用于wakeupFd_
{
int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
if (evtfd < 0)
{
LOG_SYSERR << "Failed in eventfd";
abort();
}
return evtfd;
}
EventLoop::EventLoop()
: looping_(false),
quit_(false),
callingPendingFunctors_(false),
threadId_(CurrentThread::tid()),
poller_(new Poller(this)),
timerQueue_(new TimerQueue(this)),
wakeupFd_(createEventfd()),//创建一个文件文件描述符
wakeupChannel_(new Channel(this, wakeupFd_))//创建相应的 wakeupFd_的Channel
{
LOG_TRACE << "EventLoop created " << this << " in thread " << threadId_;
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();//向poller注册Channel
}
EventLoop::~EventLoop()
{
assert(!looping_);
::close(wakeupFd_);
t_loopInThisThread = NULL;
}
void EventLoop::loop()
{
assert(!looping_);
assertInLoopThread();
looping_ = true;
quit_ = false;
while (!quit_)
{
activeChannels_.clear();
pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
for (ChannelList::iterator it = activeChannels_.begin();
it != activeChannels_.end(); ++it)
{
(*it)->handleEvent();
}
doPendingFunctors();//新增，执行任务队列
}
LOG_TRACE << "EventLoop " << this << " stop looping";
looping_ = false;
}
void EventLoop::quit()
{
quit_ = true;
if (!isInLoopThread())
{
wakeup();//新增，从poll阻塞唤醒当前线程，并结束loop循环
}
}
void EventLoop::runInLoop(const Functor& cb)//暴露给外不线程的函数
{
if (isInLoopThread())
{
cb();//在所属线程中直接执行
}
else
{
queueInLoop(cb);//加入所属的EventLoop的函数队列，并唤醒EventLoop的线程
}
}
void EventLoop::queueInLoop(const Functor& cb)//加入所属的EventLoop的函数队列，并唤醒EventLoop的线程
{
{
MutexLockGuard lock(mutex_);//上锁
pendingFunctors_.push_back(cb);
}
if (!isInLoopThread() || callingPendingFunctors_)//唤醒线程的条件中有callingPendingFunctors_，因为如果执行函数队列PendingFunctors_时有可能调用queueinloop（），!isInLoopThread()部分返回false，，无法执行wakeup（）,无法被及时的唤醒
{
wakeup();
}
}
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::updateChannel(Channel* channel)
{
assert(channel->ownerLoop() == this);
assertInLoopThread();
poller_->updateChannel(channel);
}
void EventLoop::abortNotInLoopThread()
{
LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
<< " was created in threadId_ = " << threadId_
<< ", current thread id = " << CurrentThread::tid();
}
void EventLoop::wakeup()//唤醒，对wakeupFd_文件描述符写入，poll从监听阻塞中返回事件的发生
{
uint64_t one = 1;
ssize_t n = ::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 = ::read(wakeupFd_, &one, sizeof one);
if (n != sizeof one)
{
LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
}
}
void EventLoop::doPendingFunctors()//执行函数队列，
{
std::vector<Functor> functors;
callingPendingFunctors_ = true;
{
MutexLockGuard lock(mutex_);
functors.swap(pendingFunctors_);//交换函数队列，减少锁的持有时间，并且避免了函数队列中函数有可能执行queueinloop（），造成无限制的死循环的情况。
}
for (size_t i = 0; i < functors.size(); ++i)
{
functors[i]();
}
callingPendingFunctors_ = false;
}
有了runInLoop（）函数就可以对EventLoop的公有函数进行调整，实现公有函数在不同线程调用的安全性。
代码如下，因为很简单就不注释了：
TimerId TimerQueue::addTimer(const TimerCallback& cb,
Timestamp when,
double interval)
{
Timer* timer = new Timer(cb, when, interval);
loop_->runInLoop(
boost::bind(&TimerQueue::addTimerInLoop, this, timer));
return TimerId(timer);
}
void TimerQueue::addTimerInLoop(Timer* timer)
{
loop_->assertInLoopThread();
bool earliestChanged = insert(timer);
if (earliestChanged)
{
resetTimerfd(timerfd_, timer->expiration());
}
}
class EventLoopThread
上文EventLoop已经实现了大部分的功能，但无法在线程中使用，具体以代码展示来说明：
#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(3, run4);
g_flag = 3;
}
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);
}
在该程序中loop只能在该进程中执行，无法直接放入一个新线程中执行，或者说需要每次写一个函数来执行EventLoop的创建和运行，再将这个函数放入新线程中执行，这样不如直接提供一个线程运行的class，用于在新线程中创建EventLoop的实例，并执行loop（），返回EventLoop的指针，代码如下：
#ifndef MUDUO_NET_EVENTLOOPTHREAD_H
#define MUDUO_NET_EVENTLOOPTHREAD_H
#include "thread/Condition.h"
#include "thread/Mutex.h"
#include "thread/Thread.h"
#include <boost/noncopyable.hpp>
namespace muduo
{
class EventLoop;
class EventLoopThread : boost::noncopyable
{
public:
EventLoopThread();
~EventLoopThread();
EventLoop* startLoop();
private:
void threadFunc();
EventLoop* loop_;
bool exiting_;
Thread thread_;
MutexLock mutex_;
Condition cond_;
};
}
#endif
成员函数如下：
EventLoopThread::EventLoopThread()
: loop_(NULL),
exiting_(false),
thread_(boost::bind(&EventLoopThread::threadFunc, this)),//新线程执行函数的绑定
mutex_(),
cond_(mutex_)
{
}
EventLoopThread::~EventLoopThread()
{
exiting_ = true;
loop_->quit();
thread_.join();
}
EventLoop* EventLoopThread::startLoop()//启动线程
{
assert(!thread_.started());
thread_.start();
{
MutexLockGuard lock(mutex_);
while (loop_ == NULL)
{
cond_.wait();
}
}
return loop_;
}
void EventLoopThread::threadFunc()//创建EventLoop，并运行loop
{
EventLoop loop;
{
MutexLockGuard lock(mutex_);
loop_ = &loop;
cond_.notify();
}
loop.loop();
//assert(exiting_);
}
至此，muduo网络库的基本框架已经介绍完毕，第二章网络库的实现同样是基于第一章的结构之上。。。
未完待续