六、用户态实现

????????通过上篇文章，我们可知在用户态实现IO多路复用的大规模可靠UDP服务器不能用原生的udp协议，否则会面临当链接数增加的时候CPU和内存占用都增加很快直至承受不住的情况。要规避内核的种种弊端，就必须要用用户态协议栈来实现，这也是最适合最顺理成章的方案。我们可以站在前人的肩膀上，让我们来看一下DPDP+MTCP架构。

1.dpdk+mtcp

????????dpdk在驱动层工作，前文有所介绍。mtcp在tcp协议层工作，不仅实现了mtcp_socket、mtcp_bind、mtcp_listen、mtcp_accept、mtcp_connect、mtcp_read、mtcp_write等函数（在mtcp\src\include\mtcp_api.h中定义），而且还实现了mtcp_epoll_create、mtcp_epoll_ctl、mtcp_epoll_wait函数（在mtcp\src\include\mtcp_epoll.h中定义）。

2.为mtcp增加udp支持

????????如果要给mtcp增加对udp的支持，可以参考我以前的文章《修改 mtcp 源代码来添加对 udp 支持》https://blog.csdn.net/scabo/article/details/115967931

????????另外我们不仅要上文中的mtcp支持udp的内容，还要mtcp_epoll支持udp，我们需要修改或者增加下面几个函数的相关函数mtcp_socket、mtcp_accept、mtcp_read、mtcp_write、mtcp_epoll_create、mtcp_epoll_ctl、mtcp_epoll_wait，让这些函数支持udp，由于比较简单，没有用到Linux内核的epoll，连文件描述符都没有用，所以就不一一说明了。

????????StreamHTInsert、StreamHTSearch函数实现了TCP协议的sip、sport与sock对应关系，具体的hash算法在tcp_stream.c(62):unsigned int HashFlow(const void *f)函数中。也需要对应实现一套UDP协议的相关算法（同样比较简单）。

????????mtcp的TCP协议传输数据相关算法基本在mtcp\src\include\tcp_stream.h、mtcp\src\tcp_stream.c中，看起来并没有BBR拥塞控制等较新算法。

3.为mtcp增加kcp支持

????????mtcp收包处理过程为RunMainLoop-> ProcessPacket-> ProcessIPv4Packet-> ProcessUDPPacket（此函数是在上文中描述过需要自己实现的）。ProcessUDPPacket内增加对kcp的引用，然后在RunMainLoop主函数里增加kcp的flush函数的引用就可以了。具体详见文章上面的KCP和XKCPTUN项目。

4.Linux UIO和dpdk interrupt

????????Linux的UIO是用户空间硬件I/O驱动机制，适合用该机制驱动的硬件可以映射并读写内存、控制设备产生的中断。dpdk通过igb_uio.ko模块与内核UIO交互，详见文章《DPDK—IGB_UIO，与UIO Framework 进行交互的内核模块》http://t.zoukankan.com/hzcya1995-p-13309267.html

????????另外dpdk也提供了interrupt模式，通过在线程中使用epoll模型，监听UIO设备的事件，来模拟操作系统的中断处理。在dpdk启动时外部调用rte_eal_init ->rte_eal_intr_init函数。

int rte_eal_intr_init(void)
{
??? TAILQ_INIT(&intr_sources);
??? rte_ctrl_thread_create(&intr_thread, "eal-intr-thread", NULL, eal_intr_thread_main, NULL);
}
static __attribute__((noreturn)) void *eal_intr_thread_main(__rte_unused void *arg)
{
??? for (;;) {
??????? int pfd = epoll_create(1);
??????? epoll_ctl(pfd, EPOLL_CTL_ADD, intr_pipe.readfd, &pipe_event)
??????? TAILQ_FOREACH(src, &intr_sources, next) {
??????????? epoll_ctl(pfd, EPOLL_CTL_ADD, src->intr_handle.fd, &ev)
??????? }
??????? eal_intr_handle_interrupts(pfd, numfds);
??? }
}
static void eal_intr_handle_interrupts(int pfd, unsigned totalfds)
{
??? for(;;) {
??????? nfds = epoll_wait(pfd, events, totalfds, EAL_INTR_EPOLL_WAIT_FOREVER);
?????? ?eal_intr_process_interrupts(events, nfds)
??? }
}
static int eal_intr_process_interrupts(struct epoll_event *events, int nfds)
{
??? for (n = 0; n < nfds; n++) {
??????? int r = read(intr_pipe.readfd, buf.charbuf,sizeof(buf.charbuf));
??????? TAILQ_FOREACH(cb, &src->callbacks, next) {
??????? ????active_cb.cb_fn(active_cb.cb_arg);
??????? }
??? }
}

????????在dpdk\drivers\net\路径下，有dpdk支持的网卡avf、e1000、bnx2x等的驱动。https://blog.csdn.net/Quyuan2009/article/details/50347823文章中是网卡列表。在网卡初始化的时候，会调用rte_eth_dev_init()--->eth_igb_dev_init()--->rte_intr_callback_register()注册中断处理函数，也就是调用TAILQ_INSERT_TAIL(&intr_sources, src, next);。参考文章https://blog.csdn.net/u013982161/article/details/51761330。

????????另外在dpdk\lib\librte_eal\linuxapp\eal\eal_interrupts.c中，也提供了eal_init_tls_epfd、rte_epoll_ctl、rte_epoll_wait函数来直接调用epoll相关函数。

????????由上可见，dpdk的中断模式也是调用了epoll相关函数实现，要实现IO多路复用的大规模可靠UDP还要修改内核，所以此种方案并不理想。

5.最优方案

????????前面说过在用户态下实现IO多路复用的大规模可靠UDP是较好选择，但是上面的方案似乎都有一些不顺畅的感觉。所以我认为我们最佳的方案是效仿mTcp，做一个适合于自己项目的mUdp，使用dpdk+mUdp的方案将是一个少线程的IO多路复用方案，有较少CPU和内存占用，将是最优方案。在这个方案下，我们需要自定义socket、udp_hashinfo、可能需要mmap、就绪队列（rdllist），内核线程休眠唤醒、线程与CPU亲和。我们不需要socket、文件描述符fd、红黑树。这样的好处是实现比较简单，我们还可以把主要精力放在可靠udp的协议的优化上。

大概实现流程如下：

struct sock
{
??? int socket; //索引号
??? int stat; //空闲、工作、断开等状态
}
main
{
??? 创建n个struct sock（n根据自己项目例如1000万）
??? 创建udp_hashinfo
??? mainloop {
??????? uint16_t n_rx = rte_eth_rx_burst(port, queue, pkts_burst, PKT_BURST);
??????? foreach(i in n_rx) {
??????????? result = parse(pkts_burst[i]);//通过sip,sport找到sock，如果没有返回new_accept
??????????? if(result== new_accept) {
??????????????? sock中找到最近的空闲的;
??????????????? 将sip,sport,sock_index写入udp_hashinfo；
??????????? }
??????????? else {
??????????????? 处理消息pkts_burst[i];
??????????? }
??????? }
??? }
}

????????上述过程是单线程轮询的，在大规模链接、并发（百万级）时，会造成cpu占用过高的情况，所以我们可以有一个多线程实现流程，如下：

struct sock
{
??? int socket; //索引号
??? int threadid; //线程索引
??? int cpuid; //cpu索引
??? int stat; //空闲、工作、断开等状态
??? TAILQ_HEAD(data ) head; //数据指针队列
}
thread
{
??? 创建就绪队列（rdllist）
? ??创建udp_hashinfo
??? for(;;) {
???     休眠
        被唤醒后处理就绪队列中的事件
???     轮询到超时
    }
}
mainthread
{
??? mainloop {
??????? uint16_t n_rx = rte_eth_rx_burst(port, queue, pkts_burst, PKT_BURST);
??????? foreach(i in n_rx) {
??????????? result = parse(pkts_burst[i]);//通过sip,sport找到sock，如果没有返回new_ accept
??????????? if(result== new_accept) {
??????????????? sock中找到最近的空闲的;
??????????????? 将sip,sport,sock_index写入udp_hashinfo；
??????????????? 决定用哪个appthread来处理这个链接；
??????????? }
??????????? mmap(data)
??????????? 通过sock找到线程
??????????? 将pkts_burst[i]复制到data，放入sock的head
??????????? 向线程就绪队列写入事件
??????????? 唤醒线程
??????? }
??? }
}
main
{
??? 创建n个struct sock（n根据自己项目，可以是1000万个）
??? 创建n个thread
??? 创建mainthread
??? 设置thread、mainthread的cpu亲和
}

6.链接安全性与断开的判断

????????由于我们是在用户态实现的协议栈，也没基本有建立链接的过程，所以每个数据包的合法性检查也应该根据项目自身特点在数据处理用户层实现，对于需要拉黑的ip，需要自行通知前端防火墙，或者直接丢弃。可以在开始的几个包建立消息加密算法的握手机制，如果后续消息算法错误，可以认为链接损坏断开，从而回收sock。对于应用层，还可以加上心跳，通过心跳超时来判断链接是否需要关闭回收。

????????对于安全性要求较高的应用（例如需要TLS/SSL），建立合法性链接可能是个稍微复杂一点的过程，那么还需要类似于tcp_fastopen类似的udp_fastopen快速重连机制。

????????总结上述内容，我们有了一个mudp的实现方案，实现起来并不复杂，也应该满足可以同时接入百万甚至千万链接，和百万级通讯并发。这会让我们把更多的时间和精力放在可靠性udp协议优化本身上。