前言
Transport类包装了Reactor的worker线程运行所需的所有接口，Reactor的主要功能包括执行epoll循环等待Http请求，而Transport提供了调用用户定义的Handler的接口，这样来看，如果Handler对应的是HTTP协议，Transport就是恰似TCP协议了，这是我对于为啥叫Transport的原因。
Transport类所提供的功能包括：
- 新连接到达时，Listener类调用Transport的handleNewPeer，将peer添加到Transport的peersQueue中，同时处理peersQueue，将新连接的client-fd添加到Reactor的epoll中
- HTTP请求到达时，读入数据，然后将数据交给Handler类进行解析，进而调用用户的处理函数
- 实现异步写操作
- 实现对线程的资源统计
- 实现定时机制
- 对于长连接，实现超时断开，并清除连接
Aio::Handler
Transport类的基类是Aio::Handler，这就为什么我一直在其他的帖子中混用handler和transport，并称其为Reactor的处理函数。
成员变量
-
Reactor* reactor_
这个就不用说了吧，详见Pistache源码分析 —— Reactor类 -
Context context_
用于标志transport所属的线程，主要字段就是线程的tid
struct Context
{
friend class SyncImpl;
Context(): tid(){ }
std::thread::id thread() const { return tid; }
private:
std::thread::id tid;
};
-
Reactor::Key key_
根据Reactor的设计，每个reactor_可以拥有多个不同的transport，这里的key值，就是reactor_为没个不同的transport打上的标签，通俗的讲，就是reactor_的transport数组索引。
成员函数
Aio::Handler主要定义了两个虚函数：
-
virtual void onReady(const FdSet& fds) = 0;
当reactor_的epool监听到事件时，比如HTTP请求到达，会调用此函数，然后根据到达的事件类型调用相应的处理函数。 -
virtual void registerPoller(Polling::Epoll& poller) = 0;
前面提到的各个功能(2,4,5,6,7)都是利用epool+eventfd机制实现的，因此registerPoller就是将eventfd注册到epoll上
Tcp::Transport : Aio::Handler
成员变量&&相关成员(按前言中的功能顺序)
关于PollableQueue，参阅Pistache源码分析 —— PollableQueue类
-
处理新连接
-
PollableQueue<PeerEntry> peersQueue;<private | var> -
void handleNewPeer(const std::shared_ptr<Peer>& peer);<public | func> -
void handlePeerQueue();<private | func> -
void handlePeer(const std::shared_ptr<Peer>& peer);<private | func>
-
-
处理http请求
-
std::shared_ptr<Tcp::Handler> handler_;<private | var> -
void handleIncoming(const std::shared_ptr<Peer>& peer);<private | func>
-
以上两个部分，参阅Pistache源码分析 —— Server的初始化和请求处理中第二节：处理请求
-
异步写机制
-
PollableQueue<WriteEntry> writesQueue;<private | var> -
std::unordered_map<Fd, std::deque<WriteEntry>> toWrite;<private | var> -
Lock toWriteLock;<private | var>
因为添加一条toWrite的数据的操作，是在handleNewPeer()中执行的(见Pistache源码分析 —— Server的初始化和请求处理的2.3小节)，而其他操作都是在工作线程中执行的，因此存在多线程访问，需要加锁保护 -
template <typename Buf> Async::Promise<ssize_t> asyncWrite(Fd fd, const Buf& buffer, int flags = 0)<public | func> -
void handleWriteQueue(bool flush = false);<private | func> -
void asyncWriteImpl(Fd fd);<private | func> -
ssize_t sendRawBuffer(Fd fd, const char* buffer, size_t len, int flags);<private | func>
封装了send()操作，参阅linux手册翻译——send(2) -
ssize_t sendFile(Fd fd, Fd file, off_t offset, size_t len);<private | func>
封装了sendfile()，参阅linux手册翻译——sendfile(2)
-
其余部分，参阅Pistache源码分析 —— 异步写机制
-
线程资源统计
-
Async::Deferred<rusage> loadRequest_;<private | var> -
NotifyFd notifier;<private | var> -
Async::Promise<rusage> load()<public | func> -
void handleNotify();<private | func>
-
-
定时机制
-
PollableQueue<TimerEntry> timersQueue;<private | var> -
std::unordered_map<Fd, TimerEntry> timers;<private | var> -
void armTimer(Fd fd, Duration timeout, Async::Deferred<uint64_t> deferred)<public | func> -
void disarmTimer(Fd fd);<public | func> -
void handleTimerQueue();<private | func> -
void handleTimer(TimerEntry entry);<private | func> -
void armTimerMs(Fd fd, std::chrono::milliseconds value, Async::Deferred<uint64_t> deferred);<private | func> -
void armTimerMsImpl(TimerEntry entry);<private | func>
-
上面这两部分，分别是基于rusage和timerfd机制实现的，可以参考linux手册翻译——timerfd_create(2)和linux手册翻译——getrusage(2)，我也没时间展开分析了。
-
断开连接
-
std::unordered_map<Fd, std::shared_ptr<Peer>> peers;<protected | var>
保存新连接的client-fd及其peer对象 -
void removePeer(const std::shared_ptr<Peer>& peer);<protected | func>
由 handlePeerDisconnection()调用，主要操作包括:- 将新连接从
peers中移除 - 将clientf-fd从
toWrite中移除 - 将clientf-fd从reactor的epoll中移除
- close(client-fd)
- 将新连接从
-
void handlePeerDisconnection(const std::shared_ptr<Peer>& peer);<private | func>
当执行handleIncoming()接受Http请求数据出现错误时执行，主要是调用removePeer()，以及handler_的onDisconnection(peer)函数，详见Pistache源码分析 —— Handler类。
-
-
重载父类
-
void registerPoller(Polling::Epoll& poller) override;<public | func>
以下分别对应了1、3、4、5功能，他们都是基于eventfd机制实现的，在这里添加到reactor的epoll中
-
void Transport::registerPoller(Polling::Epoll& poller)
{
writesQueue.bind(poller);
timersQueue.bind(poller);
peersQueue.bind(poller);
notifier.bind(poller);
}
-
void onReady(const Aio::FdSet& fds) override;<public | func>
当reactor的epoll接受到事件之后，最终就会调用到onReady()，然后根据fd以及事件类型，调用相应的处理函数。
void Transport::onReady(const Aio::FdSet& fds)
{
for (const auto& entry : fds)
{
if (entry.getTag() == writesQueue.tag())
{
handleWriteQueue();
}
else if (entry.getTag() == timersQueue.tag())
{
handleTimerQueue();
}
else if (entry.getTag() == peersQueue.tag())
{
handlePeerQueue();
}
else if (entry.getTag() == notifier.tag())
{
handleNotify();
}
else if (entry.isReadable())
{
auto tag = entry.getTag();
if (isPeerFd(tag))
{
auto& peer = getPeer(tag);
handleIncoming(peer);
}
else if (isTimerFd(tag))
{
auto it = timers.find(static_cast<decltype(timers)::key_type>(tag.value()));
auto& entry_ = it->second;
handleTimer(std::move(entry_));
timers.erase(it->first);
}
}
else if (entry.isWritable())
{
auto tag = entry.getTag();
auto fd = static_cast<Fd>(tag.value());
{
Guard guard(toWriteLock);
auto it = toWrite.find(fd);
if (it == std::end(toWrite))
{
throw std::runtime_error(
"Assertion Error: could not find write data");
}
}
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
// Try to drain the queue
asyncWriteImpl(fd);
}
}
}
-
std::shared_ptr<Aio::Handler> clone() const override;<public | func>
创建一个克隆的transport对象，以分配给每个线程，详见Pistache源码分析 —— Reactor类
-
其他
-
void init(const std::shared_ptr<Tcp::Handler>& handler);<public | func>
-
void Transport::init(const std::shared_ptr<Tcp::Handler>& handler)
{
handler_ = handler;
handler_->associateTransport(this);
}
-
void flush();<public | func>
此函数与stream相关，我没关注这部分
网友评论