netty4 总结

November 24 2016

话说，最近，要在内部推广 mainstay3 所以，需要先深度分享下 netty
以便让大家能放心使用 mainstay3 with netty.

这里总结在自己blog里面, netty的版本为 4.1.xx。

EventloopGroup

对于开发者来说，主要关心紫色圈出部分。其余的都已经封装完毕

有几个关键点。

Eventloopsize 建议设置为 2 的次方，dispatch 使用位移，更快。
侦听一个端口，只会绑定到 BossEventLoopGroup 中的一个 Eventloop，所以， BossEventLoopGroup 配置多个也无用。

EventLoop

只使用 tcp 和异步阻塞的话主要关心以下2个 EventLoop

NioEventLoop - 基于java 原生nio

level-triggered （水平触发）

EpollEventLoop - native jni 直接调用 epoll, only work on linux

edge-triggered （边缘触发）更少的系统调用
C代码，更少GC，更少synchronized
暴露了更多的Socket配置参数

流程图

关键点

整个loop 干的事情就是 select -> processIO -> runAllTask
这是一个死循环
那么这个loop 如何自己优雅退出？ noway，只能通过外部添加 CloseTask，比如添加到 MpscQueue
deadline 为定时任务的触发时间，避免 select 阻塞, 让定时任务不能及时执行。
在select 这一步解决 epollbug

关于解决 epoll bug的原理是应当 “阻塞”的 select 变得不再阻塞。
所以只需要统计下 select 次数就行了

部分关键代码:

for(;;){
    int selectedKeys = selector.select(timeoutMillis); // select with timeout
    selectCnt ++;
    // 我由于 select 阻塞 而等待了 timeoutMillis 毫秒， 说明， 我阻塞了，说明没有bug
    if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
        selectCnt = 1;
    } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
            selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
        // 在小于 timeoutMillis 毫秒的时间内 select 的次数超过了 阀值(512) 次
        rebuildSelector();
        selector = this.selector;

        selector.selectNow();// Select again
        selectCnt = 1;
        break;
    }
}

ChannelPipeline

读写链，运行于 eventloop 之内, 内部是一个双向链表。

以下面的代码举例子。

public void initChannel(SocketChannel ch) throws Exception {
  ch.pipeline()
      .addLast(new StringDecoder())
      .addLast(new StringEncoder())
      .addLast(new EchoHandler());
}

从上面可以看出几个点

对于 pipeline.fireChannelRead 这种 fire 开头的API 的请求从 HeadContext 开始找链的next Handler 且继承了 ChannelInboundHandler 的 Handler
对于 pipeline.write (还有 bind, connect, flush 这种 非fire 开头) 的请求相反从 TailContext 开始找链的 perv Handler 且继承了 ChannelOutboundHandler 的 Handler
对于内部直接 ctx.xxx 自然不是从 tail ／ head 开始，路径更短，更快

ByteBufPool

netty 的 ByteBufPool 是 jemelloc 的精简版，大体流程是类似的。都有一个 threadlocal 的 cache，以及公有的 Arena。

从cache 取内存无锁，从 Arena取内存有锁。所以一般建议将 Arena 的数量设置为 EventLoop 数量，以达到无竞争。

关系图如下:

关于参数的设定参看 io.netty.buffer.PooledByteBufAllocator 这个类.

如果先不管 Arena 是什么的话，Cache 是如何缓存的？

给每个申请的内存块的大小都分配一个Queue (一样大小的都recycle 都放回一个 Queue)
如果给每个大小都定义个Queue 反而浪费内存了，所以可申请的内存块大小是指定规格的。
对申请的内存根据预定义的规格向上取整
所有的规格都放在一个数组里面，寻找起来 O1 的速度。

根据默认参数的话规格如下:

1
2
3

tiny:   [16 32 48 64 80 ... 496]
small:  [512, 1024, 2048, 4096]
normal: [8192x2^0,  8192x2^1, 8192x2^2 .. 8192x2^11]

举个例子：当申请的内存为 10个byte 则向上取整 16 ，分配16byte，取 tiny[0] 里取内存即可

Arena 为分配器，是预申请内存块的管理者。 jemelloc 的初衷为了减少内存碎片，那么如何减少？

当申请比如 1 byte 的话就会预申请一大块内存，然后分配其中一部分给 1byte。

为什么是分配一部分？不然内存管理花费的内存开销将大于内存实际的开销。。

Arena 是一个分配器，内部管理的内存块为 Chunk, 申请大于 Chunk 大小的内存将直接向系统申请。

Chunk 内部由 n 个 page 组成， page 是一个虚拟的概念，因为 page 还可以切分为 n 个 SubPage

他们的关系图如下:

ps: 左侧为默认参数

这里说一下Arena 需要做的事情

首先, 一个 Arena 用 ChunkList 管理着 n 个 Chunk，一个Chunk 默认 16mb, 当一个 Chunk 分配不了时，新建一个新的 Chunk

Chunk & ChunkList

Arena 内部还对 Chunk 做了使用量的监测。

Arena 新建了 6个 ChunkList 分别为 qInit, q000, q025, q050, q075, q100

她们依次组成一个双向链表，分别又一个最小使用量和最大使用量的阀值

//          PoolChunkList<T> nextList, int minUsage, int maxUsage, int chunkSize
q100 = new PoolChunkList<T>(null, 100, Integer.MAX_VALUE, chunkSize);
q075 = new PoolChunkList<T>(q100, 75, 100, chunkSize);
q050 = new PoolChunkList<T>(q075, 50, 100, chunkSize);
q025 = new PoolChunkList<T>(q050, 25, 75, chunkSize);
q000 = new PoolChunkList<T>(q025, 1, 50, chunkSize);
qInit = new PoolChunkList<T>(q000, Integer.MIN_VALUE, 25, chunkSize);

然后在每次 allocate 和 deallocate 时判断 Chunk 是否触及阀值，然后分别上移/下移到合适的 ChunkList

为了减小内存碎片，分配的优先级如下:

q050 > q025 > q000 > qInit > q075

代码如下

private synchronized void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
    if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) ||
        q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) ||
        q075.allocate(buf, reqCapacity, normCapacity)) {
        ++allocationsNormal;
        return;
    }
    // ...

为什么这么弄？

首先, 可以分配的内存时这样的，回到之前的规格。对于 8192x2^11 byte 相当于 100% 的 chunk 了。

而比他小的分别为 50%, 25%, 12.5%, 6.25%, 3.125% …

所以如果将优先级的第一第二换个位置则可能造成大量的 75% 。。。

Page & SubPage

Chunk 由 2048 个 Page 组成。

一个 Page 默认 8k。

Chunk 里面将这 2048 个 page 用其 2 倍的空间 byte[4096] memoryMap 以完全二叉树管理。

深度为 11 的共 2048 个元素表示该 page 有没有被分配

深度 1-10 的节点都是索引

每个节点的 value 默认为该节点在树的深度。

当 page 被分配掉时，则将自己的 value ＋ 1 也就是变成12

每次被分配后都要同时更新自己的 parent 的 value, parent 的 value 取 2 个 child 中小的一个。

private void updateParentsAlloc(int id) {
    while (id > 1) {
        int parentId = id >>> 1;
        byte val1 = value(id);
        byte val2 = value(id ^ 1);
        byte val = val1 < val2 ? val1 : val2;
        setValue(parentId, val);
        id = parentId;
    }
}

比如当要分配 8k 的大小时直接找第 11 层找第一个 val 为 11 的(unused), 找到后更新自己以及parent节点。

这样当比如需要分配 8MB 内存是，立即找 depth 为 1 的2个节点，只要她的 value 不等于 depth 就知道该节点的 child 被分配过了。

然后 Subpage 的作用是用来标示分配小于一个 page 的内存块。

但是 Subpage 不再是用二叉树管理了。而是用更方便的方式。

一个 Subpage 只能分配一种 size 大小，然后用 3个long 共 8x8x8 个bit 标示是否被使用。

比如如果 page 为 8k ，由于共 512 个标示可用，所以最小分配单元为 16 byte.

但是如果最小分配单元为 32 byte，则共只需要 256 个标示即可。(3个 long 不需要全部用完)

Subpage 也是一个链表, 相同规格的subpage 链在一起。

然后 PoolArena 初始化了 n 个size 的 Subpage 的链表头, 并用来维护。

1 2	private final PoolSubpage<T>[] tinySubpagePools; private final PoolSubpage<T>[] smallSubpagePools;

每次新创建 Subpage 会吧 subpage 链到相应的 idx 的 PoolSubpage 内。

private long allocateSubpage(int normCapacity) {
    // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.
    // This is need as we may add it back and so alter the linked-list structure.
    PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);
    synchronized (head) {
        int d = maxOrder; // subpages are only be allocated from pages i.e., leaves
        int id = allocateNode(d);
        if (id < 0) {
            return id;
        }

        final PoolSubpage<T>[] subpages = this.subpages;
        final int pageSize = this.pageSize;

        freeBytes -= pageSize;

        int subpageIdx = subpageIdx(id);
        PoolSubpage<T> subpage = subpages[subpageIdx];
        if (subpage == null) {
            subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);
            subpages[subpageIdx] = subpage;
        } else {
            subpage.init(head, normCapacity);
        }
        return subpage.allocate();
    }
}