Skip to content

netty connection_state

Jump to bottom
wangkaiwu edited this page Apr 14, 2022 · 7 revisions

netty connection 状态

监听新的连接

public final class NioEventLoop extends SingleThreadEventLoop {
    private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        int readyOps = k.readyOps();
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
        }
    }
}

java 的 accept 方法

private final class NioMessageUnsafe extends AbstractNioUnsafe {
    private final List<Object> readBuf = new ArrayList<Object>();

    @Override
    public void read() {
        int localRead = doReadMessages(readBuf);
        // 这里的 ChannelRead 是作用于 ServerChannel(监听的channel) 的 handler
        pipeline.fireChannelRead(readBuf.get(0));
    }

    @Override
    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = SocketUtils.accept(javaChannel());
        buf.add(new NioSocketChannel(this, ch));
    }
}
public final class SocketUtils {
    public static SocketChannel accept(final ServerSocketChannel serverSocketChannel) throws IOException {
        return serverSocketChannel.accept();
    }
}
class ServerSocketChannelImpl extends ServerSocketChannel implements SelChImpl {
    private final ReentrantLock acceptLock = new ReentrantLock();

    @Override
    public SocketChannel accept() throws IOException {
        acceptLock.lock();
        FileDescriptor newfd = new FileDescriptor();
        InetSocketAddress[] isaa = new InetSocketAddress[1];
//        int n = accept(this.fd, newfd, isaa);
        int n = accept0(this.fd, newfd, isaa);
        InetSocketAddress isa = isaa[0];
        SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);
        acceptLock.unlock();
        return sc;
    }
}

jvm 的 accept 方法

JNIEXPORT jint JNICALL
Java_sun_nio_ch_ServerSocketChannelImpl_accept0(JNIEnv *env, jobject this,
                                                jobject ssfdo, jobject newfdo,
                                                jobjectArray isaa)
{
    jint ssfd = (*env)->GetIntField(env, ssfdo, fd_fdID);
    SOCKETADDRESS sa;
    socklen_t sa_len = sizeof(SOCKETADDRESS);
    newfd = accept(ssfd, &sa.sa, &sa_len);
    (*env)->SetIntField(env, newfdo, fd_fdID, newfd);
    remote_ia = NET_SockaddrToInetAddress(env, &sa, (int *)&remote_port);
    isa = (*env)->NewObject(env, isa_class, isa_ctorID, remote_ia, remote_port);
    (*env)->SetObjectArrayElement(env, isaa, 0, isa);
    return 1;
}

glibc 的 accept 方法

int
__libc_accept (int fd, __SOCKADDR_ARG addr, socklen_t *len)
{
    return SYSCALL_CANCEL (accept, fd, addr.__sockaddr__, len);
}

linux 内核的 accept 方法

SYSCALL_DEFINE3(accept, int, fd, struct sockaddr __user *, upeer_sockaddr,
		int __user *, upeer_addrlen)
{
	return sys_accept4(fd, upeer_sockaddr, upeer_addrlen, 0);
}
SYSCALL_DEFINE4(accept4, int, fd, struct sockaddr __user *, upeer_sockaddr,
		int __user *, upeer_addrlen, int, flags)
{
    struct socket *sock, *newsock;
    struct file *newfile;
    int err, newfd, fput_needed;
    
    sock = sockfd_lookup_light(fd, &err, &fput_needed);
    newsock = sock_alloc();
    newsock->type = sock->type;
    newsock->ops = sock->ops;
    newfd = sock_alloc_file(newsock, &newfile, flags);
    err = security_socket_accept(sock, newsock);
    fput(newfile);
    put_unused_fd(newfd);
    fput_light(sock->file, fput_needed);
    return err;
}

ServerChannel 的 channelRead

public class DefaultChannelPipeline implements ChannelPipeline {
    final AbstractChannelHandlerContext head;
    final AbstractChannelHandlerContext tail;

    protected DefaultChannelPipeline(Channel channel) {
        // 由此可知，head 和 tail 就是两个空实现
        tail = new TailContext(this);
        head = new HeadContext(this);
        head.next = tail;
        tail.prev = head;
    }

    @Override
    public final ChannelPipeline fireChannelRead(Object msg) {
        // pipeline 是管道的意思，这里是 fireChannelRead 这个管道的起始
        // 这里的 msg 是 SocketChannel
        AbstractChannelHandlerContext.invokeChannelRead(head, msg);
        return this;
    }
}

ServerChannel 的 pipeline 中 head 的 ChannelRead

abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        // 当前的 next 实际上是 head
        // 这里返回的是 ServerChannel 绑定的 EventLoop
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) { // 这个 executor 是不是当前线程，默认是
            // 如果为 ServerChannel 添加了一个自定义 Handler，那么这里的 next 就是那个 Handler。
            // 例如本例中的 LoggingHandler 
            next.invokeChannelRead(m);
        }
    }

    private void invokeChannelRead(Object msg) {
        // 当前还在 head 对象的方法体内
        if (invokeHandler()) {
            // head 是个 AbstractChannelHandlerContext，它关联了一个唯一的 handler。
            // 而且 head 的 handler 就是自身
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        }
    }
}
final class HeadContext extends AbstractChannelHandlerContext
        implements ChannelOutboundHandler, ChannelInboundHandler {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 当前执行的是 head 的 channelRead
        // head 只是简单的转发到下一个 context
        ctx.fireChannelRead(msg);
    }
}

ServerChannel 的 pipeline 中 head 执行下一个 context

abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    @Override
    public ChannelHandlerContext fireChannelRead(final Object msg) {
        // 当前仍然处于 head 对象中
        // findContextInbound 会查到下一个 context 是 LoggingHandler 
        invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
        return this;
    }

    private AbstractChannelHandlerContext findContextInbound(int mask) {
        AbstractChannelHandlerContext ctx = this;
        // currentExecutor当前仍然是serverChannel 对应的 executor
        EventExecutor currentExecutor = executor();
        do {
            // pipeline 中的下一个 context
            ctx = ctx.next;
            // 校验一下这个 context 是否满足 mask 的要求
        } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_INBOUND));
        // 查到了合适的 context
        return ctx;
    }

    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        // next 对应的是 LoggingHandler
        EventExecutor executor = next.executor();
        // executor 仍然是 serverChannel 绑定的 eventLoop
        if (executor.inEventLoop()) {
            // 执行 LoggingHandler 的 read 
            next.invokeChannelRead(m);
        }
    }

    private void invokeChannelRead(Object msg) {
        // 当前已经处于 LoggingHandler 的 context 中
        if (invokeHandler()) {
            // handler() 返回的就是 LoggingHandler
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        }
    }
}

ServerChannel 的 pipeline 中 Logging Context 执行 channelRead

public class LoggingHandler extends ChannelDuplexHandler {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (logger.isEnabled(internalLevel)) {
            logger.log(internalLevel, format(ctx, "READ", msg));
        }
        // 转到 pipeline 的下一个 context
        ctx.fireChannelRead(msg);
    }
}

ServerChannel 的 pipeline 中 Logging Context 转发到下一个 context

abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    @Override
    public ChannelHandlerContext fireChannelRead(final Object msg) {
        // findContextInbound 此时会找到 ServerBootstrapAcceptor，这是 pipeline 的最后一环
        invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
        return this;
    }

    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        // next 对应的是 ServerBootstrapAcceptor
        EventExecutor executor = next.executor();
        // executor 还是 serverChannel 的 eventLoop
        if (executor.inEventLoop()) {
            next.invokeChannelRead(m);
        }
    }

    private void invokeChannelRead(Object msg) {
        // 此时已经处于 ServerBootstrapAcceptor 的方法中
        if (invokeHandler()) {
            // handler() 对应的就是 ServerBootstrapAcceptor 
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        }
    }

}

ServerChannel 的 pipeline 中 ServerBootstrap Context 执行 channelRead

private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        final Channel child = (Channel) msg;
        // 把 socketChannel 的 pipeline 置入 childHandler
        child.pipeline().addLast(childHandler);
        // 设置属性
        setChannelOptions(child, childOptions, logger);
        setAttributes(child, childAttrs);
        // 将 channel 注册到一个 worker 的 EventLoop
        childGroup.register(child);
    }
}

注册一个 SocketChannel

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    @Override
    public ChannelFuture register(Channel channel) {
        // 通过 next() 选择一个 eventLoop
        return next().register(channel);
    }
}

选择一个 eventLoop，并与 SocketChannel 绑定

public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
    @Override
    public ChannelFuture register(Channel channel) {
        // 将 channel 注册到选中的 eventLoop 中
        return register(new DefaultChannelPromise(channel, this));
    }

    @Override
    public ChannelFuture register(final ChannelPromise promise) {
        promise.channel().unsafe().register(this, promise);
        return promise;
    }
}

执行 SocketChannel 注册，并执行后续处理

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    protected abstract class AbstractUnsafe implements Unsafe {
        private boolean neverRegistered = true;

        @Override
        public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            AbstractChannel.this.eventLoop = eventLoop;
            if (eventLoop.inEventLoop()) {
            } else {
                eventLoop.execute(new Runnable() {
                    @Override
                    public void run() {
                        register0(promise);
                    }
                });
            }
        }

        private void register0(ChannelPromise promise) {
            boolean firstRegistration = neverRegistered;
            doRegister();
            neverRegistered = false;
            pipeline.invokeHandlerAddedIfNeeded();
            // 以下是 channel 完整的生命周期
            // channelRegistered -> channelActive -> channelInactive -> channelUnregistered
            pipeline.fireChannelRegistered();
            if (isActive()) {
                if (firstRegistration) {
                    // channelActive 最终执行的是当前 channel 的 handler 中的 channelActive
                    pipeline.fireChannelActive();
                }
            }
        }
    }
}
public abstract class AbstractNioChannel extends AbstractChannel {
    volatile SelectionKey selectionKey;

    @Override
    protected void doRegister() throws Exception {
        // javaChannel() 对应的是刚刚建立连接的 channel。eventLoop() 就是选择的 eventLoop
        // 所以这里的意思就是把 channel 注册到 eventLoop 对应的 selector 上，并返回注册信息
        selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
    }
}

执行 SocketChannel 的 handlerAdded

public class DefaultChannelPipeline implements ChannelPipeline {
    private boolean firstRegistration = true;
    private boolean registered;

    final void invokeHandlerAddedIfNeeded() {
        if (firstRegistration) {
            firstRegistration = false;
            callHandlerAddedForAllHandlers();
        }
    }

    private void callHandlerAddedForAllHandlers() {
        final PendingHandlerCallback pendingHandlerCallbackHead;
        synchronized (this) {
            registered = true;
            pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
            this.pendingHandlerCallbackHead = null;
        }
        PendingHandlerCallback task = pendingHandlerCallbackHead;
        while (task != null) {
            task.execute();
            task = task.next;
        }
    }
}
public class DefaultChannelPipeline implements ChannelPipeline {
    private final class PendingHandlerAddedTask extends PendingHandlerCallback {
        @Override
        void execute() {
            EventExecutor executor = ctx.executor();
            if (executor.inEventLoop()) {
                callHandlerAdded0(ctx);
            }
        }
    }

    private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
        ctx.callHandlerAdded();
    }
}
abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    final void callHandlerAdded() throws Exception {
        if (setAddComplete()) {
            handler().handlerAdded(this);
        }
    }
}

执行 SocketChannel 的 initChannel 逻辑

@Sharable
public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter {
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isRegistered()) {
            if (initChannel(ctx)) {
            }
        }
    }

    private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
        if (initMap.add(ctx)) {
            initChannel((C) ctx.channel());
        }
    }
}
public final class EchoServer {
    public static void main(String[] args) throws Exception {
        final EchoServerHandler serverHandler = new EchoServerHandler();
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        if (sslCtx != null) {
                            p.addLast(sslCtx.newHandler(ch.alloc()));
                        }
                        p.addLast(new LoggingHandler(LogLevel.INFO));
                        p.addLast(serverHandler);
                    }
                });
    }
}

连接被动关闭触发

public final class NioEventLoop extends SingleThreadEventLoop {
    private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        int readyOps = k.readyOps();
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
        }
    }
}
protected class NioByteUnsafe extends AbstractNioUnsafe {
    @Override
    public final void read() {
        final ChannelPipeline pipeline = pipeline();
        ByteBuf byteBuf = null;
        boolean close = false;
        try {
            byteBuf = allocHandle.allocate(allocator);
            allocHandle.lastBytesRead(doReadBytes(byteBuf));
            if (allocHandle.lastBytesRead() <= 0) {
                close = allocHandle.lastBytesRead() < 0;
            }
            if (close) {
                // 正常被动关闭
                closeOnRead(pipeline);
            }
        } catch (Throwable t) {
            // 异常被动关闭
            handleReadException(pipeline, byteBuf, t, close, allocHandle);
        }
    }
}

被动关闭的触发条件

public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
    @Override
    protected int doReadBytes(ByteBuf byteBuf) throws Exception {
        final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
        allocHandle.attemptedBytesRead(byteBuf.writableBytes());
        return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
    }
}
public abstract class AbstractByteBuf extends ByteBuf {
    int writerIndex;

    @Override
    public int writeBytes(ScatteringByteChannel in, int length) throws IOException {
        int writtenBytes = setBytes(writerIndex, in, length);
        if (writtenBytes > 0) {
            writerIndex += writtenBytes;
        }
        return writtenBytes;
    }
}
abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    @Override
    public final int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {
        return in.read(internalNioBuffer(index, length));
    }
}
class SocketChannelImpl extends SocketChannel implements SelChImpl {
    private final ReentrantLock readLock = new ReentrantLock();

    @Override
    public int read(ByteBuffer buf) throws IOException {
        int n = 0;
        readLock.lock();
        n = IOUtil.read(fd, buf, -1, nd);
        readLock.unlock();
        return IOStatus.normalize(n);
    }
}
public class IOUtil {
    static int read(FileDescriptor fd, ByteBuffer dst, long position,
                    NativeDispatcher nd) throws IOException {
        return read(fd, dst, position, false, -1, nd);
    }

    static int read(FileDescriptor fd, ByteBuffer dst, long position,
                    boolean directIO, int alignment, NativeDispatcher nd) throws IOException {
        if (dst instanceof DirectBuffer)
            return readIntoNativeBuffer(fd, dst, position, directIO, alignment, nd);
    }

    private static int readIntoNativeBuffer(FileDescriptor fd, ByteBuffer bb,
                                            long position, boolean directIO,
                                            int alignment, NativeDispatcher nd) throws IOException {
        int pos = bb.position();
        int lim = bb.limit();
        int rem = (pos <= lim ? lim - pos : 0);
        if (position != -1) {
        } else {
            n = nd.read(fd, ((DirectBuffer) bb).address() + pos, rem);
        }
        return n;
    }
}
class SocketDispatcher extends NativeDispatcher {
    int read(FileDescriptor fd, long address, int len) throws IOException {
        return FileDispatcherImpl.read0(fd, address, len);
    }
}

Jvm read0 实现

JNIEXPORT jint JNICALL
Java_sun_nio_ch_FileDispatcherImpl_read0(JNIEnv *env, jclass clazz,
                             jobject fdo, jlong address, jint len) {
    jint fd = fdval(env, fdo);
    void *buf = (void *)jlong_to_ptr(address);
    return convertReturnVal(env, read(fd, buf, len), JNI_TRUE);
}

Glibc read 实现

__fortify_function __wur ssize_t
read (int __fd, void *__buf, size_t __nbytes)
{
  if (__bos0 (__buf) != (size_t) -1)
    {
      if (!__builtin_constant_p (__nbytes))
	return __read_chk (__fd, __buf, __nbytes, __bos0 (__buf));

      if (__nbytes > __bos0 (__buf))
	return __read_chk_warn (__fd, __buf, __nbytes, __bos0 (__buf));
    }
  return __read_alias (__fd, __buf, __nbytes);
}
ssize_t
__read_chk (int fd, void *buf, size_t nbytes, size_t buflen) {
    if (nbytes > buflen)
        __chk_fail ();
    return INLINE_SYSCALL (read, 3, fd, buf, nbytes);
}
// 由于会把 val 转成无符号数，所以 -1 >= val >= -4095 都会返回 1，其余都返回 0
# define INTERNAL_SYSCALL_ERROR_P(val, err) \
  ((unsigned long int) (long int) (val) >= -4095L)

# define INLINE_SYSCALL(name, nr, args...) \
  ({									      \
    // 系统调用 read，resultvar 是返回值
    unsigned long int resultvar = INTERNAL_SYSCALL (name, , nr, args);	      \
    if (__glibc_unlikely (INTERNAL_SYSCALL_ERROR_P (resultvar, )))	      \
      {									      \
	__set_errno (INTERNAL_SYSCALL_ERRNO (resultvar, ));		      \
	
	resultvar = (unsigned long int) -1;				      \
      }									      \
    (long int) resultvar; })

linux read 实现

SYSCALL_DEFINE3(read, unsigned int, fd, char __user *, buf, size_t, count)
{
	struct file *file;
	ssize_t ret = -EBADF;
	int fput_needed;

	file = fget_light(fd, &fput_needed);
	if (file) {
		loff_t pos = file_pos_read(file);
		ret = vfs_read(file, buf, count, &pos);
		file_pos_write(file, pos);
		fput_light(file, fput_needed);
	}

	return ret;
}
ssize_t vfs_read(struct file *file, char __user *buf, size_t count, loff_t *pos)
{
	ssize_t ret;

	if (!(file->f_mode & FMODE_READ))
		return -EBADF;
	if (!file->f_op || (!file->f_op->read && !file->f_op->aio_read))
		return -EINVAL;
	if (unlikely(!access_ok(VERIFY_WRITE, buf, count)))
		return -EFAULT;

	ret = rw_verify_area(READ, file, pos, count);
	if (ret >= 0) {
		count = ret;
		if (file->f_op->read)
			ret = file->f_op->read(file, buf, count, pos);
		else
			ret = do_sync_read(file, buf, count, pos);
		if (ret > 0) {
			fsnotify_access(file);
			add_rchar(current, ret);
		}
		inc_syscr(current);
	}

	return ret;
}

当 linux 系统调用 read 返回的时候，不但返回值是 -1，同时还设置了线程的 errno 值，且该值是 104。详见 /include/asm-generic/errno.h

Jvm read0 抛出异常

jint
convertReturnVal(JNIEnv *env, jint n, jboolean reading)
{
    if (n > 0) /* Number of bytes written */
        return n;
    else if (n == 0) {
        if (reading) {
            return IOS_EOF; /* EOF is -1 in javaland */
        } else {
            return 0;
        }
    }
    else if (errno == EAGAIN || errno == EWOULDBLOCK)
        return IOS_UNAVAILABLE;
    else if (errno == EINTR)
        return IOS_INTERRUPTED;
    else {
        const char *msg = reading ? "Read failed" : "Write failed";
        JNU_ThrowIOExceptionWithLastError(env, msg);
        return IOS_THROWN;
    }
}
JNIEXPORT void JNICALL
JNU_ThrowIOExceptionWithLastError(JNIEnv *env, const char *defaultDetail)
{
    JNU_ThrowByNameWithLastError(env, "java/io/IOException", defaultDetail);
}
JNIEXPORT void JNICALL
JNU_ThrowByNameWithLastError(JNIEnv *env, const char *name,
                             const char *defaultDetail)
{
    char buf[256];
    size_t n = getLastErrorString(buf, sizeof(buf));
    if (n > 0) {
        jstring s = JNU_NewStringPlatform(env, buf);
        if (s != NULL) {
            jobject x = JNU_NewObjectByName(env, name,
                                            "(Ljava/lang/String;)V", s);
            if (x != NULL) {
                (*env)->Throw(env, x);
            }
        }
    }
}
JNIEXPORT size_t JNICALL
getLastErrorString(char *buf, size_t len)
{
    if (errno == 0 || len < 1) return 0;
    getErrorString(errno, buf, len);
    return strlen(buf);
}
JNIEXPORT int JNICALL
getErrorString(int err, char *buf, size_t len)
{
    // err == 104
    if (err == 0 || len < 1) return 0;
    return strerror_r(err, buf, len);
}

Glibc 相关调用

strerror_r 对应的就是 __xpg_strerror_r

#define	ECONNRESET	104

#ifdef ECONNRESET
_S(ECONNRESET, N_("Connection reset by peer"))

#: sysdeps/gnu/errlist.h:374
msgid "Connection reset by peer"
msgstr "连接被对方重设"

int
__xpg_strerror_r (int errnum, char *buf, size_t buflen)
{
    const char *estr = __strerror_r (errnum, buf, buflen);
}

errno 查询过程

# define errno (*__errno_location ())

int * __errno_location (void) {
    return &errno;
}
#define errno __libc_errno

连接因为异常被动关闭的处理

protected class NioByteUnsafe extends AbstractNioUnsafe {
    private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close,
                                     RecvByteBufAllocator.Handle allocHandle) {
        // 先抛出异常
        pipeline.fireExceptionCaught(cause);
        // close 为 true 的情况是，read 返回了 -1
        // OOM 和 IOE 都会强制关闭连接
        if (close || cause instanceof OutOfMemoryError || cause instanceof IOException) {
            // 调用被动关闭
            closeOnRead(pipeline);
        }
    }
}

省略 pipeline 的执行过程，直接看自定义 handler 的处理。以官方的 EchoServerHandler 为例，打印了异常并主动关闭了连接。

@Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 打印异常
        cause.printStackTrace();
        // 主动关闭
        ctx.close();
    }
}
public abstract class AbstractNioByteChannel extends AbstractNioChannel {
    private boolean inputClosedSeenErrorOnRead;

    protected class NioByteUnsafe extends AbstractNioUnsafe {
        private void closeOnRead(ChannelPipeline pipeline) {
            if (!isInputShutdown0()) {
            } else {
                inputClosedSeenErrorOnRead = true;
                pipeline.fireUserEventTriggered(ChannelInputShutdownReadComplete.INSTANCE);
            }
        }
    }
}

连接因为 read 返回 -1 关闭的处理

protected class NioByteUnsafe extends AbstractNioUnsafe {
    private void closeOnRead(ChannelPipeline pipeline) {
        if (!isInputShutdown0()) {
            if (isAllowHalfClosure(config())) {
            } else {
                close(voidPromise());
            }
        }
    }
}
protected abstract class AbstractUnsafe implements Unsafe {
    private boolean inFlush0;

    @Override
    public void close(final ChannelPromise promise) {
        ClosedChannelException closedChannelException =
                StacklessClosedChannelException.newInstance(AbstractChannel.class, "close(ChannelPromise)");
        close(promise, closedChannelException, closedChannelException, false);
    }

    private void close(final ChannelPromise promise, final Throwable cause,
                       final ClosedChannelException closeCause, final boolean notify) {
        doClose0(promise);
        if (inFlush0) {
            fireChannelInactiveAndDeregister(wasActive);
        }
    }

    private void doClose0(ChannelPromise promise) {
        doClose();
    }
}
public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
    @Override
    protected void doClose() throws Exception {
        javaChannel().close();
    }
}
public abstract class AbstractInterruptibleChannel implements Channel, InterruptibleChannel {
    private final Object closeLock = new Object();
    private volatile boolean closed;

    public final void close() throws IOException {
        synchronized (closeLock) {
            if (closed)
                return;
            closed = true;
            implCloseChannel();
        }
    }
}
public abstract class AbstractSelectableChannel
        extends SelectableChannel {
    protected final void implCloseChannel() throws IOException {
        implCloseSelectableChannel();
    }
}
class SocketChannelImpl
        extends SocketChannel
        implements SelChImpl {
    public static final int SHUT_WR = 1;

    @Override
    protected void implCloseSelectableChannel() throws IOException {
        synchronized (stateLock) {
            Net.shutdown(fd, Net.SHUT_WR);
        }
    }
}

jdk 的实现

JNIEXPORT void JNICALL
Java_sun_nio_ch_Net_shutdown(JNIEnv *env, jclass cl, jobject fdo, jint jhow)
{
    // how = SHUT_WR
    int how = (jhow == sun_nio_ch_Net_SHUT_RD) ? SHUT_RD :
        (jhow == sun_nio_ch_Net_SHUT_WR) ? SHUT_WR : SHUT_RDWR;
    // 调用关闭    
    if ((shutdown(fdval(env, fdo), how) < 0) && (errno != ENOTCONN))
        handleSocketError(env, errno);
}

Glibc 的 shutdown 实现

int
__shutdown (int fd, int how)
{
    return INLINE_SYSCALL (shutdown, 2, fd, how);
}
weak_alias (__shutdown, shutdown)

linux 的 shutdown 调用

SYSCALL_DEFINE2(shutdown, int, fd, int, how)
{
	int err, fput_needed;
	struct socket *sock;

	sock = sockfd_lookup_light(fd, &err, &fput_needed);
	if (sock != NULL) {
		err = security_socket_shutdown(sock, how);
		if (!err)
			err = sock->ops->shutdown(sock, how);
		fput_light(sock->file, fput_needed);
	}
	return err;
}

fireChannelInactiveAndDeregister

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    private volatile boolean registered;

    protected abstract class AbstractUnsafe implements Unsafe {
        private void fireChannelInactiveAndDeregister(final boolean wasActive) {
            deregister(voidPromise(), wasActive && !isActive());
        }

        private void deregister(final ChannelPromise promise, final boolean fireChannelInactive) {
            // 由于用户可以在 ChannelPipeline 中进行处理时从任何方法中调用 deregister()，
            // 因此我们需要确保稍后执行实际的取消注册操作。 
            // 这是必需的，例如，我们可能在 ByteToMessageDecoder.callDecode(...) 方法中，
            // 因此仍然尝试在旧的 EventLoop 中进行处理，
            // 而用户已经将 Channel 注册到新的 EventLoop。 
            // 如果不延迟处理，取消注册操作可能会导致不同的 EventLoop 等线程调用处理程序。
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    doDeregister();
                    // 如果之前连接是 active 状态，就执行 inactive
                    if (fireChannelInactive) {
                        pipeline.fireChannelInactive();
                    }
                    // doDeregister 的作用就是撤销注册，这里就是判断一下撤销之前是不是已经撤销过了
                    if (registered) {
                        registered = false;
                        // 如果撤销注册成功则调用 Unregistered 
                        pipeline.fireChannelUnregistered();
                    }
                }
            });
        }
    }
}

doDeregister

public abstract class AbstractNioChannel extends AbstractChannel {
    @Override
    protected void doDeregister() throws Exception {
        eventLoop().cancel(selectionKey());
    }
}
public final class NioEventLoop extends SingleThreadEventLoop {
    void cancel(SelectionKey key) {
        key.cancel();
    }
}

主动关闭连接

一个是调用 Channel 的 distconnect 或者 close, 一个是调用 Context 的 close() 方法，作用都一样。

Channel 的 close 方法

虽然有两个方法，disconnect 和 close，但是殊途重归，最终都会调用 NioSocketChannel 的 doClose 方法，但更推荐调用 close

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    @Override
    public ChannelFuture disconnect() {
        return pipeline.disconnect();
    }

    @Override
    public ChannelFuture close() {
        return pipeline.close();
    }
}
public class DefaultChannelPipeline implements ChannelPipeline {
    @Override
    public final ChannelFuture disconnect() {
        return tail.disconnect();
    }

    @Override
    public final ChannelFuture close() {
        return tail.close();
    }
}
abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    @Override
    public ChannelFuture disconnect() {
        return disconnect(newPromise());
    }

    @Override
    public ChannelFuture disconnect(final ChannelPromise promise) {
        if (!channel().metadata().hasDisconnect()) {
            return close(promise);
        }
        final AbstractChannelHandlerContext next = findContextOutbound(MASK_DISCONNECT);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeDisconnect(promise);
        }
        return promise;
    }

    private void invokeDisconnect(ChannelPromise promise) {
        if (invokeHandler()) {
            ((ChannelOutboundHandler) handler()).disconnect(this, promise);
        }
    }

    @Override
    public ChannelFuture close() {
        return close(newPromise());
    }

    @Override
    public ChannelFuture close(final ChannelPromise promise) {
        final AbstractChannelHandlerContext next = findContextOutbound(MASK_CLOSE);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeClose(promise);
        }
    }

    private void invokeClose(ChannelPromise promise) {
        if (invokeHandler()) {
            ((ChannelOutboundHandler) handler()).close(this, promise);
        }
    }
}
final class HeadContext extends AbstractChannelHandlerContext
        implements ChannelOutboundHandler, ChannelInboundHandler {
    private final Unsafe unsafe;

    @Override
    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) {
        unsafe.disconnect(promise);
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
        unsafe.close(promise);
    }
}
public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {

    protected abstract class AbstractUnsafe implements Unsafe {
        @Override
        public final void disconnect(final ChannelPromise promise) {
            boolean wasActive = isActive();
            doDisconnect();
            if (wasActive && !isActive()) {
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.fireChannelInactive();
                    }
                });
            }
            closeIfClosed();
        }

        protected final void closeIfClosed() {
            if (isOpen()) {
                return;
            }
            close(voidPromise());
        }

        @Override
        public void close(final ChannelPromise promise) {
            ClosedChannelException closedChannelException =
                    StacklessClosedChannelException.newInstance(AbstractChannel.class, "close(ChannelPromise)");
            close(promise, closedChannelException, closedChannelException, false);
        }

        private void close(final ChannelPromise promise, final Throwable cause,
                           final ClosedChannelException closeCause, final boolean notify) {
            final boolean wasActive = isActive();
            Executor closeExecutor = prepareToClose();
            if (closeExecutor != null) {
                closeExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        doClose0(promise);
                    }
                });
            }
        }

        private void doClose0(ChannelPromise promise) {
            doClose();
        }
    }
}
public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
    @Override
    protected void doDisconnect() throws Exception {
        doClose();
    }

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

Context 的 close 方法

abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    @Override
    public ChannelFuture close() {
        return close(newPromise());
    }

    @Override
    public ChannelFuture close(final ChannelPromise promise) {
        final AbstractChannelHandlerContext next = findContextOutbound(MASK_CLOSE);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeClose(promise);
        }
    }

    private void invokeClose(ChannelPromise promise) {
        if (invokeHandler()) {
            ((ChannelOutboundHandler) handler()).close(this, promise);
        }
    }
}
final class HeadContext extends AbstractChannelHandlerContext
        implements ChannelOutboundHandler, ChannelInboundHandler {
    private final Unsafe unsafe;

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
        unsafe.close(promise);
    }
}
protected abstract class AbstractUnsafe implements Unsafe {
    @Override
    public void close(final ChannelPromise promise) {
        ClosedChannelException closedChannelException =
                StacklessClosedChannelException.newInstance(AbstractChannel.class, "close(ChannelPromise)");
        close(promise, closedChannelException, closedChannelException, false);
    }
}

线程模型

CMS GC 的线程模型

G1 GC 的线程模型

G1 GC 相关

G1 垃圾收集器简介

G1 Young GC 启动条件解析

G1 GC 启动并发标记的解析

G1 Mixed GC 启动条件解析

Clone this wiki locally