4. Netty源码分析之Unsafe

public final void register(final ChannelPromise promise) {
    // 当前线程是否为Channel对应的NioEventLoop线程
    if (eventLoop.inEventLoop()) {
        // 如果是，则不存在多线程并发操作，直接注册
        register0(promise);
    } else {
        // 如果不是，说明是其他线程或用户线程发起的注册，存在并发操作，将其放进NioEventLoop任务队列中执行
        try {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            logger.warn(
                    "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                    AbstractChannel.this, t);
            closeForcibly();
            closeFuture.setClosed();
            promise.setFailure(t);
        }
    }
}

private void register0(ChannelPromise promise) {
    try {
        // 判断Channel是否打开了
        if (!ensureOpen(promise)) {
            return;
        }
        // 调用AbstractNioChannel的doRegister方法。请见 Netty源码分析-Channel
        doRegister();
        registered = true;
        promise.setSuccess();
        // 注册成功
        pipeline.fireChannelRegistered();
        if (isActive()) {
            // Channel被激活
            pipeline.fireChannelActive();
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        if (!promise.tryFailure(t)) {
            logger.warn(
                    "Tried to fail the registration promise, but it is complete already. " +
                            "Swallowing the cause of the registration failure:", t);
        }
    }
}

// AbstractNioChannel.doRegister()
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            selectionKey = javaChannel().register(eventLoop().selector, 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the SelectionKey is still cached
                // for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    if (!ensureOpen(promise)) {
        return;
    }

    // See: https://github.com/netty/netty/issues/576
    if (!PlatformDependent.isWindows() && !PlatformDependent.isRoot() &&
        Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
        localAddress instanceof InetSocketAddress &&
        !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress()) {
        // Warn a user about the fact that a non-root user can‘t receive a
        // broadcast packet on *nix if the socket is bound on non-wildcard address.
        logger.warn(
                "A non-root user can‘t receive a broadcast packet if the socket " +
                "is not bound to a wildcard address; binding to a non-wildcard " +
                "address (" + localAddress + ") anyway as requested.");
    }
    
    // 是否是激活状态
    boolean wasActive = isActive();
    try {
        doBind(localAddress);
    } catch (Throwable t) {
        promise.setFailure(t);
        closeIfClosed();
        return;
    }
    if (!wasActive && isActive()) {
        // 如果是在绑定阶段成为active状态，则将调用fireChannelActive方法放进NioEventLoop执行队列中
        invokeLater(new Runnable() {
            @Override
            public void run() {
                pipeline.fireChannelActive();
            }
        });
    }
    promise.setSuccess();
}

private void invokeLater(Runnable task) {
    eventLoop().execute(task);
}

protected void doBind(SocketAddress localAddress) throws Exception {
    javaChannel().socket().bind(localAddress);
}

protected void doBind(SocketAddress localAddress) throws Exception {
    javaChannel().socket().bind(localAddress, config.getBacklog());
}

public final void disconnect(final ChannelPromise promise) {
    boolean wasActive = isActive();
    try {
        doDisconnect();
    } catch (Throwable t) {
        promise.setFailure(t);
        closeIfClosed();
        return;
    }
    if (wasActive && !isActive()) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                pipeline.fireChannelInactive();
            }
        });
    }
    promise.setSuccess();
    closeIfClosed(); // doDisconnect() might have closed the channel
}

protected void doDisconnect() throws Exception {
    throw new UnsupportedOperationException();
}

protected void doDisconnect() throws Exception {
    doClose();
}

protected void doClose() throws Exception {
    javaChannel().close();
}

public final void close(final ChannelPromise promise) {
    // 1. 是否处于刷新状态，如果处于刷新状态说明还有消息没发出去，需要等到所有消息发完后再关闭
    // 放入队列中处理
    if (inFlush0) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                close(promise);
            }
        });
        return;
    }

    // 2. 判断关闭操作是否完成，如果已完成，则不需要重复关闭链路，设置promise成功即可
    if (closeFuture.isDone()) {
        // Closed already.
        promise.setSuccess();
        return;
    }

    // 3. 执行关闭操作，将消息发送缓冲数组置空，通知JVM回收
    boolean wasActive = isActive();
    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    this.outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer.

    try {
        // 4. 关闭链路，本质是调用javaChannel的close方法
        doClose();
        closeFuture.setClosed();
        promise.setSuccess();
    } catch (Throwable t) {
        closeFuture.setClosed();
        promise.setFailure(t);
    }

    // 5. 调用ChannelOutboundBuffer.close()释放缓冲区消息，将链路关闭通知事件放进NioEventLoop执行队列中
    try {
        outboundBuffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
        outboundBuffer.close(CLOSED_CHANNEL_EXCEPTION);
    } finally {

        if (wasActive && !isActive()) {
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    pipeline.fireChannelInactive();
                }
            });
        }
        // 6. 将Channel从多路复用器上取消注册
        deregister();
    }
}

protected void doDeregister() throws Exception {
    eventLoop().cancel(selectionKey());
}

// 实际上就是将SelectionKey对应的Channel从多路复用器上去取消注册
void cancel(SelectionKey key) {
    key.cancel();
    cancelledKeys ++;
    if (cancelledKeys >= CLEANUP_INTERVAL) {
        cancelledKeys = 0;
        needsToSelectAgain = true;
    }
}

public void write(Object msg, ChannelPromise promise) {
    if (!isActive()) {
        // 未激活，TCP链路还没建立成功，根据Channel打开情况设置不同的异常
        if (isOpen()) {
            promise.tryFailure(NOT_YET_CONNECTED_EXCEPTION);
        } else {
            promise.tryFailure(CLOSED_CHANNEL_EXCEPTION);
        }
        // 无法发送，释放msg对象
        ReferenceCountUtil.release(msg);
    } else {
        // 链路状态正常，将数据和promise放进发送缓冲区
        outboundBuffer.addMessage(msg, promise);
    }
}

public void flush() {
    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    if (outboundBuffer == null) {
        return;
    }
    // 先将unflush指针修改为tail，标识本次发送的范围
    outboundBuffer.addFlush();
    flush0();
}

protected void flush0() {
    if (inFlush0) {
        // Avoid re-entrance
        return;
    }

    final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    if (outboundBuffer == null || outboundBuffer.isEmpty()) {
        return;
    }

    inFlush0 = true;

    // Mark all pending write requests as failure if the channel is inactive.
    if (!isActive()) {
        try {
            if (isOpen()) {
                outboundBuffer.failFlushed(NOT_YET_CONNECTED_EXCEPTION);
            } else {
                outboundBuffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
            }
        } finally {
            inFlush0 = false;
        }
        return;
    }

    try {
        // 调用NioSocketChannel的write方法
        doWrite(outboundBuffer);
    } catch (Throwable t) {
        outboundBuffer.failFlushed(t);
    } finally {
        inFlush0 = false;
    }
}

public final void connect(final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    // 设置不可取消 && Channel是打开状态
    if (!promise.setUncancellable() || !ensureOpen(promise)) {
        return;
    }

    try {
        if (connectPromise != null) {
            // 已经有一个连接正在处理，直接抛异常
            throw new ConnectionPendingException();
        }

        boolean wasActive = isActive();
        
        // doConnect方法具体看NioSocketChannel.doConnect()实现
        if (doConnect(remoteAddress, localAddress)) {
            // 连接成功，进行连接后操作
            fulfillConnectPromise(promise, wasActive);
        } else {
            // 连接失败，TCP无应答，结果暂未知晓
            connectPromise = promise;
            requestedRemoteAddress = remoteAddress;

            // Schedule connect timeout.
            int connectTimeoutMillis = config().getConnectTimeoutMillis();
            if (connectTimeoutMillis > 0) {
                connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                    @Override
                    public void run() {
                        ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                        ConnectTimeoutException cause =
                                new ConnectTimeoutException("connection timed out: " + remoteAddress);
                        if (connectPromise != null && connectPromise.tryFailure(cause)) {
                            close(voidPromise());
                        }
                    }
                }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
            }

            promise.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isCancelled()) {
                        if (connectTimeoutFuture != null) {
                            connectTimeoutFuture.cancel(false);
                        }
                        connectPromise = null;
                        close(voidPromise());
                    }
                }
            });
        }
    } catch (Throwable t) {
        promise.tryFailure(annotateConnectException(t, remoteAddress));
        closeIfClosed();
    }
}

private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
    if (promise == null) {
        // Closed via cancellation and the promise has been notified already.
        return;
    }

    // 判断当前激活状态
    boolean active = isActive();

    // 如果用户取消了连接，则返回false，需调用close方法关闭链路
    boolean promiseSet = promise.trySuccess();

    // 如果doConnect之前未激活，doConnect之后激活了，需要调用fireChannelActive(即使被取消了也应该调)
    if (!wasActive && active) {
        pipeline().fireChannelActive();
    }

    // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
    if (!promiseSet) {
        close(voidPromise());
    }
}

public final void finishConnect() {
    // Note this method is invoked by the event loop only if the connection attempt was
    // neither cancelled nor timed out.

    assert eventLoop().inEventLoop();

    try {
        boolean wasActive = isActive();
        // 通过javaChannel的finishConnect方法判断连接结果（如果连接失败则抛出Error，会走到catch块里）
        doFinishConnect();
        // 连接成功方法：fulfillConnectPromise(ChannelPromise promise, boolean wasActive)
        fulfillConnectPromise(connectPromise, wasActive);
    } catch (Throwable t) {
        // 关闭链路方法：fulfillConnectPromise(ChannelPromise promise, Throwable cause)
        fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
    } finally {
        // 如果连接超时时仍然没有收到服务端应答，则由定时任务关闭客户端连接，将SocketChannel从多路复用器上删除
        if (connectTimeoutFuture != null) {
            connectTimeoutFuture.cancel(false);
        }
        connectPromise = null;
    }
}

public final void read() {
            final ChannelConfig config = config();
            if (shouldBreakReadReady(config)) {
                clearReadPending();
                return;
            }
            final ChannelPipeline pipeline = pipeline();
            final ByteBufAllocator allocator = config.getAllocator();
            final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
            allocHandle.reset(config);

            ByteBuf byteBuf = null;
            boolean close = false;
...

public RecvByteBufAllocator.Handle recvBufAllocHandle() {
　　if (recvHandle == null) {
　　　　recvHandle = config().getRecvByteBufAllocator().newHandle();
　　}
　　return recvHandle;
}

static final int DEFAULT_MINIMUM = 64;//最小缓冲区长度
static final int DEFAULT_INITIAL = 1024;//初始容量
static final int DEFAULT_MAXIMUM = 65536;//最大容量

private static final int INDEX_INCREMENT = 4;//动态调整扩张步进索引
private static final int INDEX_DECREMENT = 1;//动态调整收缩步进索引

private static final int[] SIZE_TABLE;//长度向量表，数组的每个值对应一个Buffer容量

// 初始化长度向量表
// 当容量小于512时，由于缓冲区已经比较小，需要降低步进值，容量每次下调幅度降低
// 当容量大于512时，说明需要解码的消息码流比较大，需要采用调大步进幅度的方式降低动态扩张频率
static {
    List<Integer> sizeTable = new ArrayList<Integer>();
    for (int i = 16; i < 512; i += 16) {
        sizeTable.add(i);
    }

    for (int i = 512; i > 0; i <<= 1) {
        sizeTable.add(i);
    }

    SIZE_TABLE = new int[sizeTable.size()];
    for (int i = 0; i < SIZE_TABLE.length; i ++) {
        SIZE_TABLE[i] = sizeTable.get(i);
    }
}

private static int getSizeTableIndex(final int size) {
    for (int low = 0, high = SIZE_TABLE.length - 1;;) {
        if (high < low) {
            return low;
        }
        if (high == low) {
            return high;
        }

        int mid = low + high >>> 1;
        int a = SIZE_TABLE[mid];
        int b = SIZE_TABLE[mid + 1];
        if (size > b) {
            low = mid + 1;
        } else if (size < a) {
            high = mid - 1;
        } else if (size == a) {
            return mid;
        } else {
            return mid + 1;
        }
    }
}

private final int minIndex;            //最小索引
private final int maxIndex;            //最大索引
private int index;                    //当前索引
private int nextReceiveBufferSize;    //下一次预分配的Buffer大小
private boolean decreaseNow;        //是否立即执行容量收缩操作

private void record(int actualReadBytes) {
    if (actualReadBytes <= SIZE_TABLE[max(0, index - INDEX_DECREMENT - 1)]) {
        if (decreaseNow) {
            index = max(index - INDEX_DECREMENT, minIndex);
            nextReceiveBufferSize = SIZE_TABLE[index];
            decreaseNow = false;
        } else {
            decreaseNow = true;
        }
    } else if (actualReadBytes >= nextReceiveBufferSize) {
        index = min(index + INDEX_INCREMENT, maxIndex);
        nextReceiveBufferSize = SIZE_TABLE[index];
        decreaseNow = false;
    }
}

public final void read() {
        final ChannelConfig config = config();
        if (shouldBreakReadReady(config)) {
            clearReadPending();
            return;
        }
        final ChannelPipeline pipeline = pipeline();
        final ByteBufAllocator allocator = config.getAllocator();
        final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
        allocHandle.reset(config);

        ByteBuf byteBuf = null;
        boolean close = false;
        try {
            do {
                // 通过接收缓冲区分配器计算下次预分配的缓冲区容量并创建ByteBuffer
                byteBuf = allocHandle.allocate(allocator);
                // 这里分两步：1. doReadBytes(byteBuf)：调用NioSocketChannel.doReadBytes(..)，返回本次读取的字节数（返回0-无消息   返回小于0-发生了IO异常）
                // 2. 设置lastBytesRead，用以下面的处理
                allocHandle.lastBytesRead(doReadBytes(byteBuf));
                if (allocHandle.lastBytesRead() <= 0) {
                    // 走到这里说明上一步没有读取到数据，释放ByteBuffer
                    byteBuf.release();
                    byteBuf = null;
                    close = allocHandle.lastBytesRead() < 0;
                    if (close) {
                        // 发生了IO异常，需关闭连接
                        readPending = false;
                    }
                    break;
                }

                allocHandle.incMessagesRead(1);
                readPending = false;
                // 一次读操作，触发一次read事件
                pipeline.fireChannelRead(byteBuf);
                byteBuf = null;
            } while (allocHandle.continueReading());
            
            allocHandle.readComplete();
            // 触发读操作结束事件
            pipeline.fireChannelReadComplete();

            if (close) {
                closeOnRead(pipeline);
            }
        } catch (Throwable t) {
            handleReadException(pipeline, byteBuf, t, close, allocHandle);
        } finally {
            // Check if there is a readPending which was not processed yet.
            // This could be for two reasons:
            // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
            // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
            //
            // See https://github.com/netty/netty/issues/2254
            if (!readPending && !config.isAutoRead()) {
                removeReadOp();
            }
        }
    }
}



protected int doReadBytes(ByteBuf byteBuf) throws Exception {
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.attemptedBytesRead(byteBuf.writableBytes());
    return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}

public int writeBytes(ScatteringByteChannel in, int length) throws IOException {
    ensureWritable(length);
    int writtenBytes = setBytes(writerIndex, in, length);
    if (writtenBytes > 0) {
        writerIndex += writtenBytes;
    }
    return writtenBytes;
}