From e305b6af5e1a7c58b521186a637b2831e1e9fde1 Mon Sep 17 00:00:00 2001 From: Idel Pivnitskiy Date: Thu, 6 Aug 2020 04:54:46 -0700 Subject: [PATCH] Create a new h2 stream only if the parent channel is still active (#10444) Motivation: If a request to open a new h2 stream was made from outside of the EventLoop it will be scheduled for future execution on the EventLoop. However, during the time before the `open0` task will be executed the parent channel may already be closed. As the result, `Http2MultiplexHandler#newOutboundStream()` will throw an `IllegalStateException` with the message that is hard to interpret correctly for this use-case: "Http2FrameCodec not found. Has the handler been added to a pipeline?". Modifications: - Check that the parent h2 `Channel` is still active before creating a new stream when `open0` task is picked up by EventLoop; Result: Users see a correct `ClosedChannelException` in case the parent h2 `Channel` was closed concurrently with a request for a new stream. --- .../http2/Http2StreamChannelBootstrap.java | 8 +- .../Http2StreamChannelBootstrapTest.java | 118 +++++++++++++++++- 2 files changed, 124 insertions(+), 2 deletions(-) diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrap.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrap.java index bd8eb98a9d..a5c9ca901d 100644 --- a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrap.java +++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrap.java @@ -122,7 +122,13 @@ public final class Http2StreamChannelBootstrap { open0(ctx, promise); } else { final ChannelHandlerContext finalCtx = ctx; - executor.execute(() -> open0(finalCtx, promise)); + executor.execute(() -> { + if (channel.isActive()) { + open0(finalCtx, promise); + } else { + promise.setFailure(new ClosedChannelException()); + } + }); } } catch (Throwable cause) { promise.setFailure(cause); diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrapTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrapTest.java index e941fc5586..ee2d2b7832 100644 --- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrapTest.java +++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2StreamChannelBootstrapTest.java @@ -15,22 +15,138 @@ */ package io.netty.handler.codec.http2; +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultithreadEventLoopGroup; +import io.netty.channel.local.LocalAddress; +import io.netty.channel.local.LocalChannel; +import io.netty.channel.local.LocalHandler; +import io.netty.channel.local.LocalServerChannel; import io.netty.util.concurrent.DefaultPromise; import io.netty.util.concurrent.EventExecutor; import io.netty.util.concurrent.Promise; +import io.netty.util.internal.logging.InternalLogger; +import io.netty.util.internal.logging.InternalLoggerFactory; +import org.hamcrest.core.IsInstanceOf; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.ExpectedException; -import static org.hamcrest.CoreMatchers.is; +import java.nio.channels.ClosedChannelException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; + +import static io.netty.handler.codec.http2.Http2FrameCodecBuilder.forClient; +import static io.netty.handler.codec.http2.Http2FrameCodecBuilder.forServer; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; public class Http2StreamChannelBootstrapTest { + private static final InternalLogger logger = + InternalLoggerFactory.getInstance(Http2StreamChannelBootstrapTest.class); + + @Rule + public ExpectedException exceptionRule = ExpectedException.none(); + + private volatile Channel serverConnectedChannel; + + @Test + public void testStreamIsNotCreatedIfParentConnectionIsClosedConcurrently() throws Exception { + EventLoopGroup group = null; + Channel serverChannel = null; + Channel clientChannel = null; + try { + final CountDownLatch serverChannelLatch = new CountDownLatch(1); + group = new MultithreadEventLoopGroup(LocalHandler.newFactory()); + LocalAddress serverAddress = new LocalAddress(getClass().getName()); + ServerBootstrap sb = new ServerBootstrap() + .channel(LocalServerChannel.class) + .group(group) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(Channel ch) { + serverConnectedChannel = ch; + ch.pipeline().addLast(forServer().build(), newMultiplexedHandler()); + serverChannelLatch.countDown(); + } + }); + serverChannel = sb.bind(serverAddress).sync().channel(); + + Bootstrap cb = new Bootstrap() + .channel(LocalChannel.class) + .group(group) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(Channel ch) { + ch.pipeline().addLast(forClient().build(), newMultiplexedHandler()); + } + }); + clientChannel = cb.connect(serverAddress).sync().channel(); + assertTrue(serverChannelLatch.await(3, SECONDS)); + + final CountDownLatch closeLatch = new CountDownLatch(1); + final Channel clientChannelToClose = clientChannel; + group.execute(new Runnable() { + @Override + public void run() { + try { + closeLatch.await(); + clientChannelToClose.close().syncUninterruptibly(); + } catch (InterruptedException e) { + logger.error(e); + } + } + }); + + Http2StreamChannelBootstrap bootstrap = new Http2StreamChannelBootstrap(clientChannel); + Promise promise = clientChannel.eventLoop().newPromise(); + bootstrap.open(promise); + assertThat(promise.isDone(), is(false)); + closeLatch.countDown(); + + exceptionRule.expect(ExecutionException.class); + exceptionRule.expectCause(IsInstanceOf.instanceOf(ClosedChannelException.class)); + promise.get(3, SECONDS); + } finally { + safeClose(clientChannel); + safeClose(serverConnectedChannel); + safeClose(serverChannel); + if (group != null) { + group.shutdownGracefully(0, 3, SECONDS); + } + } + } + + private static Http2MultiplexHandler newMultiplexedHandler() { + return new Http2MultiplexHandler(new ChannelInitializer() { + @Override + protected void initChannel(Http2StreamChannel ch) { + // noop + } + }); + } + + private static void safeClose(Channel channel) { + if (channel != null) { + try { + channel.close().syncUninterruptibly(); + } catch (Exception e) { + logger.error(e); + } + } + } + @Test public void open0FailsPromiseOnHttp2MultiplexHandlerError() { Http2StreamChannelBootstrap bootstrap = new Http2StreamChannelBootstrap(mock(Channel.class));