light0x00 / lighty Goto Github PK

动机

目前 Lighty 会返回 “完整能力的” ListenableFutureTask 对象给用户, 这会导致用户可以随意调用

• ListenableFutureTask#setSuccess
• ListenableFutureTask#setFailure
• ListenableFutureTask#run

这些方法被调用后, 意味着 future 已完成, 进而导致 listener 会在错误的时机被执行.

方案

返回给用户的 ListenableFutureTask 对象应当限制这些能力, 可加一层装饰器实现.

具体的需要加以限制的 future 包括:

• bindFuture , 由 ServerBootstrap#bind 返回
• connectableFuture , 由 ClientBootstrap#connect 返回
• closedFuture , 由 NioSocketChannel#closedFuture , NioChannel#close, 返回
• connectedFuture , 由 NioSocketChannel#connectedFuture 返回
• writeFuture , 由 NioSocketChannel#write 返回
• shutdownFuture , 由 NioSocketChannel#shutdownOutput, NioSocketChannel#shutdownInput 返回

动机

应用层协议有超时失败需求, 比如

• 客户端建立连接希望在指定时间内未成功则失败
• 一个 Http 请求希望在指定时间如果未收到响应则失败

目前 Lighty 对于这两种情况均未支持.

连接超时解决方案

读超时解决方案

问题

问题 1, 如何关联请求与响应? 即每当收到一个响应如何知道其对应哪一个请求?

在同一个连接上发出的多次请求, 一定是先发出的请求先收到响应的, 只需要预先在发出请求时就维护请求顺序, 那么在收到响应时就可以对应起来.

问题 2, 如何处理“姗姗来迟”的响应?

可以维护一个计数值 timeoutCount, 当延迟任务里检测到响应超时, 那么就意味着接下来会有一个响应一定是 ”姗姗来迟“ 的(或者一直不来), 这时 timeoutCount += 1, 当之后的某个时间点(也可以是无期限)这个响应来了, 那么丢弃掉, timeoutCount -= 1

每当 channel 读事件发生, 要分配内存装载就绪的数据时, 对缓冲区大小的动态统计和预测能力.

• 比如一开始每次分配 1kb 的读缓冲区, 但是每次都需要分几次才能读完 channel, 那么意味着缓冲区分配小了, 下一次应该分配更大的缓冲区.
• 而另一面, 如果每次装载就绪数据时, 缓冲区都有很多剩余空间, 则意味着缓冲区大了, 下次应该分配更小的.

实现此功能的目的在于, 减少系统调用的次数, 提升性能.
这种弹性能力, 能更好的适配应用层协议的特征. 比如大文件传输协议, 往往数据总是大批量的到来. 而 rpc 协议, 数据总是小段的到来.

如题

动机

目前用户写入的数据暂存在输出队列 (OutputBuffer) 中, writeable 事件时, nio线程遍历这个队列, 将数据“一个接着一个” 写入 socket.

存在的问题是, 这些数据很可能“小块的”, 以 Linux 的默认 recive buffer size 8KB 为例, 写入总大小 8KB 数据, 如果分为小块, 每次 1KB 则需要 8 次 write 系统调用 , 而如果借助 Scatter/gather I/O 采用合并写, 则只需一次系统调用, 写入 8KB.

the default size of the receive buffer for a TCP
socket. This value overwrites the initial default
buffer size from the generic global
net.core.rmem_default defined for all protocols.
The default value is 87380 bytes. (On Linux 2.4,
this will be lowered to 43689 in low-memory
systems.
参考:
关于 receive buffer 默认值 tcp_rmem
关于 Scatter/gather I/O

方案

**: 使输出缓冲队列( OutputBuffer ) 具有合并 buffer 的能力

如何判断是否应当合并? 这需要知道底层 socket 的 send buffer size, 每当用户通过 write 投入数据时, 检查队列中的“上一个数据” 是否足够大(相对于 send buffer size) , 不够大则合并.

怎么合并?

用户写入的数据可以是 byte[] 、ByteBuffer、 RingBuffer 、FileChannel, 其中前 3 个都是可以使用统一的 ByteBuffer 来表示的, 合并后使用 ByteBuffer[] 表示.

这里需要特别注意的一点是 RingBuffer, 其可读出数据可能分布为两段:

    writePosition   readPosition
          │              │
  ┌───────▼──────────────▼──────────┐
  │ unread│              │  unread  │
  └───────┴──────────────┴──────────┘

可使用如下方法生成两个分片

//class ByteBuffer
public abstract ByteBuffer slice(int index, int length)

在得到 ByteBuffer[] 后, 调用 java 中的 Scatter/gather I/O 接口, 写入数据到 socket

//GatheringByteChannel
public long write(ByteBuffer[] srcs, int offset, int length)

还有一点需要注意, 由于 Lighty 中, 写入是异步的, 用户每次写入都会返回一个 ListenableFutureTask , 在数据实际被写入 socekt 后, 需要由这个 future 回调用户的 listener.

未合并前, 每次写入都有一个唯一的 future, 而合并后则对应多个. 合并过程中, 必须要维护 buffer 与 future 的映射关系, 可能一对一, 可能多对一(RingBuffer 的情况), 每次合并写后都需要根据实际写入的字节数来判定 ByteBuffer[] 中哪个写完了, 哪个没写完, 写完的就调用对应的 future.

最后还有一点, 按照 Lighty 的隐式约定, conetxt.write() 写入的如果是 RecycleBuffer 类型, 则会在实际写入 socket 后被自动回收, 这个事可以通过 future.addListener 来做.

目前的实现用户每次 write 都会唤醒一次 nio 线程, 应该改为只有 flush 或者 write 的数据量足够多才唤醒.