首先向云风致敬, 他的 skynet 给了我很多启发. skynet 的核心是多线程环境下的消息管理, 如何让消息安全高效的从一个服务传递到另一个服务. 服务在线程池中运行.
skynet 实现了一个轻巧而高效的网络模块. 我用c++重写了这个部分, 并做了一些修改使它结构更清晰, 更方便使用. 这篇文章描述的就是这个网络模块的实现. 它基于 skynet v1.0.0-alpha10 (2015-8-17).
我尽量不贴代码, 只用文字来描述.
概述
网络模块内部不会启动新的线程, 它在程序的主线程中运行. 由于网络io通常不是性能的热点, 一个线程是可行的. 需要注意的是, 如果所有功能都在一个线程处理, 会有一些隐患(下文提到), 通常我们需要为网络层单起一个线程.
网络模块提供几个接口, 用来监听网络地址, 发起连接, 断开连接. 发送数据. 这些接口都是非阻塞的. 同时它们也是线程安全的, 可以用多个线程来调用这些接口.
网络模块提供一个消息泵, 弹出网络层的状态和收到的数据. 用户可以根据具体情境, 在主线程处理这些数据, 也可以用新的工作线程来处理.
Epoll
使用 epoll 来做多路复用IO接口. epoll 使用默认的水平触发模式. 用 Epoll 类来封装 epoll 的操作.
一个 Epoll 对象只有一个数据成员, 就是 epoll fd. 它有这几个接口:
创建, 销毁, 删除fd
添加需要监控的 fd: 添加的同时监听这个 fd 的读事件.
修改某个 fd 需要监控的事件类型: 通常我们总是需要监听读事件, 读事件根据参数来判断是否需要监听.
获取事件信息: 返回一个事件数组, 每个数组成员是一个结构体, 结构体包含这样的信息: 哪个fd发生了读/写事件.
简单的 Socket
用一个Socket类来封装 socket fd. 每个连接对应一个 socket 对象.
socket 对象必然包含一个 fd. 由于我们要求非阻塞的写数据, 所以socket对象还拥有一个 WriteBuffer. 这是最简单的情况.
SocketManager
在实际中, 网络连接会频繁的建立或断开. 不停的实例化然后又销毁Socket对象, 显然是低效的.
一个改进的方法是将不需要的 socket 对象放到池中, 需要的时候, 从池中拿出来, 初始化内部数据后继续使用.
这个方法使 socket 对象的数量逐渐增加, 然后稳定在某个数量.
另一方法是一开始就静态实例化一个足够大的socket对象数组. 数组的大小可以很简单的估算出来. 我们通过浪费一些内存换来了稳定的结构.
我采用的是第二种方法.
SocketManager的实现很简单. 它提供一个接口来返回一个未被使用的socket.
需要注意的是, 这个接口必须是线程安全的.
改进的 Socket
为了支持 SocketManager, 简单的Socket类需要一些额外的信息.
id: 用来区分其它socket对象, 实际上它等于该对象在数组中的下标.
state: 是否被使用的状态, 实际上还有许多其它状态
用户数据: 用户可以在里面保存一些信息. 比如该连接属于哪个模块(服务).
Socket 的写缓存
这个缓存用来暂时记录需要发送的数据, 在 fd 可写时, 发送它们.
为了避免字符串拷贝的开销, 我们约定, 应用层负责为需要发送的数据分配内存, 网络层的缓存只保存传递过来的指针. 数据发送完成后, 网络层释放这块内存. 这是一个重要的设计.
缓存内部数据结构是一个先进先出的队列(链表实现). 每条数据对应一个节点.
每个节点记录三个数据: 数据地址, 数据长度, 原始数据地址
发送数据的流程是这样的:
应用层需要发送数据, socket id, buffer, size
网络层找到这个 socket, 如果它的写缓存是空的, 直接发送(非阻塞的).
如果数据全部发送完, 操作完成.
如果数据没有发送完, 或者缓存不是空的, 把 buffer, size, offset(可能一部分数据已经发送了)记录到缓存中.
网络层异步的, 在 fd 可写的时候, 发送头节点的的数据, 根据发送的大小, 维护偏移.
如果某个节点的数据都发送完了, 释放这个节点的 buffer. 然后删除这个节点.
Socket 的读缓存
skynet并没有为socket实现读缓存. 每当socket收到数据, 网络层的消息泵就把这些数据传递出去.
这是因为网络层不涉及逻辑, 它不知道如何解析数据. 正如发送数据时, 网络层只是发送一个内存块, 它不知道内存块里是什么数据, 这些数据又是如何组织.
让逻辑层来解析和处理数据, 是一种适用面更广的设计.
用管道来同步
根实际情境, 你可以让把服务器的所有任务都在一个线程里处理. 但是更多的时候, 为了减低耦合, 利用多CPU, 网络模块在一个独立的线程中运行. 而网络的使用者, 可能来自多个线程.
所以, 我们需要同步应用层对网络层的操作. 一个线程安全的消息队列可以很好的完成这个任务.
还有一个更简洁的办法, 那就是使用管道.
用管道来作为应用层和网络层的桥梁. 使用者向管道写入操作, 管道保证操作同步, 网络线程从管道读出操作并执行. 管道大大减少了代码的复杂度.
网络层的消息指令非常简短, 绝大多数指令只有十几个字节. 发送数据的指令传递的是数据指针, 非常短小.
管道保证, 一次往管道里写入4096字节以内的数据, 都是原子操作. 我们不用烦心考虑同步问题.
另一方面, 管道的缓存空间是64K. 由于网络层能快速的处理指令, 这个容量够用了.
考虑最坏的情况, 管道缓存满了, 那么写管道的操作会阻塞. 等到管道里的指令被处理, 腾出空间, 数据再被写入.
所以指令不会丢失.
考虑一种特殊情况: 写管道操作和网络线程是同一线程, 并且管道满了. 由于写操作被阻塞, 网络线程无法处理指令, 系统就陷入死锁.
所以, 应用层和网络层最好是属于不同的线程.
监听流程
我用监听操作的逻辑流程来举例说明应用层和网络层是如何配合的.
以use开头的操作属于应用线程(代码可能在网络层), 用net开头的操作属于网络线程.
use. 调用Listen函数.
use. 创建socket fd, 绑定网络地址, 开始监听.
use. 将一个状态为free的Socket对象互斥的置为reserve.
use. 把一条监听指令写入管道, 指令参数包括 socket fd, Socket id.
use. Listen函数返回 Socket id.
net. 从管道中读出监听指令.
net. 根据 Socket id, 找到socket对象, 初始化它, 把它的状态置为plisten.
use. 调用Start函数, 传入 Listen函数返回的Socket id
use. 启动指令写入管道.
net. 从管道读出启动指令.
net. 找到Socket对象, 把socket fd 添加到 epoll. Socket状态置为listen.
net. 处理连接事件.
流程结束.
Socket 的状态
socket的初始状态时free, 表明它未被使用.
如果要使用socket, 状态从 free 变为 reserve, 表明它已经被预定了, 做什么还不知道.
如果要用某个socket来监听, 状态从 reserve 变为 plisten, 此时它没有被添加到epoll .
系统初始化完成后, 网络启动, socket fd 添加到 epoll, 状态从 plisten 变为 listen. 它可以处理连接请求了.
如果有连接请求, 把 一个socket 的状态从 free 变为 reserve, 再变为 paccept, 此时它没有被添加到epoll .
如果应用层决定接受这个连接, 把 socket fd 添加到 epoll, 状态从 paccept 变为 connected. 它可以发送和接收数据了.
如果要用某个socket来发起连接. 把 一个socket 的状态从 free 变为 reserve, 发起连接并把fd 添加进epoll.
状态是 connecting, 连接成功后(三次握手完成), 状态变为 connected. 它可以发送和接收数据了.
如果要关闭一个socket, 缓存里有数据, 状态先设为 halfclose, 数据发送完后, 设为 free.
状态时 halfclose的socket不会再读数据, 但是会继续把缓存的数据发送完.
消息泵
好像水泵从水池里抽水一样, 应用层循环调用MsgLoop函数, 从网络层提取信息.
如果循环代码在一个独立的线程里执行, 这个线程就是网络线程.
消息泵会抛出这些消息:
data: socket 收到数据, 返回数据指针和长度, 应用层使用完后要释放指针指向的内存.
close: 连接关闭
open: 开始监听或连接建立.
accept: 有连接请求.
exit: 网络层退出.
异常处理
发起连接时, 我们将socket设置为非阻塞的, 再调用 connect方法, 此时会立即返回一个错误码.
如果错误码是 EINPROGRESS, 说明连接正在进行(三次握手). 我们把 fd 添加到 epoll 中. 当连接完成的时候, epoll 会发出一个事件.
如果是其它错误码, 说明连接失败了. 终止连接操作.
往管道读写数据的时候, 如果返回 EINTR, 说明操作被系统中断, 需要重新写.
如果是其它错误, 打印一条错误日志.
往socket写数据的时候, 如果返回 EINTR(被中断), 可以马上重写,
如果返回 EAGAIN(被阻塞), 过一会再写.
如果是其它错误, 打印一条错误日志, 然后断开连接.
往socket读数据的时候, 如果返回 EINTR(被中断), 重新读,
如果返回 EAGAIN(被阻塞), 打印一条错误日志, 重新读.
如果是其它错误, 断开连接.
网络常量
这些常量可以根据具体情境进行调整.
创建epoll的时候, 要告诉内核监听的数目有多大, 对应的 EPOLL_FD_MAX = 1024;
socket 监听网络地址时, 需要指定半连接队列的大小, 对应的 LISTEN_BACKLOG = 32;
从socket fd 读数据时, 由于我们事先不知道会读出多少数据, 在动态申请内存时, 一开始申请的大小是一个常量: MIN_READ_BUFFER
= 64; 每个socket的read_buffer_size是独立的, 而且会根据前一条数据的大小来动态调整: 增大一倍或缩小一倍.
由于Socket对象是被静态初始化的, 它们的数量要大于 EPOLL_FD_MAX , 而且我们希望可以快速的找到一个 free状态的socket对象. 所以它们的数量很多: MAX_SOCKET
= 65536;