Skip to content

Commit d079dec

Browse files
zrlwzrlw
authored
[3.3] Add HTTP/2 server connection preface process (#15535)
* Add HTTP/2 server connection preface process * Close channel handler context and release resources if client connection preface does not arrive in time
1 parent e0a59bd commit d079dec

4 files changed

Lines changed: 198 additions & 12 deletions

File tree

‎dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyHttp2FrameCodec.java‎

Lines changed: 107 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
*/
1717
package org.apache.dubbo.remoting.http12.netty4.h2;
1818

19+
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
20+
import org.apache.dubbo.common.logger.LoggerFactory;
1921
import org.apache.dubbo.remoting.http12.h2.Http2Header;
2022
import org.apache.dubbo.remoting.http12.h2.Http2InputMessage;
2123
import org.apache.dubbo.remoting.http12.h2.Http2InputMessageFrame;
@@ -25,6 +27,9 @@
2527
import org.apache.dubbo.remoting.http12.netty4.NettyHttpHeaders;
2628

2729
import java.io.OutputStream;
30+
import java.util.LinkedList;
31+
import java.util.List;
32+
import java.util.concurrent.TimeUnit;
2833

2934
import io.netty.buffer.ByteBuf;
3035
import io.netty.buffer.ByteBufInputStream;
@@ -37,9 +42,32 @@
3742
import io.netty.handler.codec.http2.Http2DataFrame;
3843
import io.netty.handler.codec.http2.Http2Headers;
3944
import io.netty.handler.codec.http2.Http2HeadersFrame;
45+
import io.netty.util.concurrent.ScheduledFuture;
46+
47+
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_TIMEOUT_SERVER;
4048

4149
public class NettyHttp2FrameCodec extends ChannelDuplexHandler {
4250

51+
private static final ErrorTypeAwareLogger LOGGER =
52+
LoggerFactory.getErrorTypeAwareLogger(NettyHttp2FrameCodec.class);
53+
54+
private static final long SETTINGS_FRAME_ARRIVAL_TIMEOUT = 3;
55+
56+
private final NettyHttp2SettingsHandler nettyHttp2SettingsHandler;
57+
58+
private final List<CachedMsg> cachedMsgList = new LinkedList<>();
59+
60+
private boolean settingsFrameArrived;
61+
62+
private ScheduledFuture<?> settingsFrameArrivalTimeoutFuture;
63+
64+
public NettyHttp2FrameCodec(NettyHttp2SettingsHandler nettyHttp2SettingsHandler) {
65+
this.nettyHttp2SettingsHandler = nettyHttp2SettingsHandler;
66+
if (!nettyHttp2SettingsHandler.subscribeSettingsFrameArrival(this)) {
67+
settingsFrameArrived = true;
68+
}
69+
}
70+
4371
@Override
4472
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
4573
if (msg instanceof Http2HeadersFrame) {
@@ -53,15 +81,76 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
5381

5482
@Override
5583
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
56-
if (msg instanceof Http2Header) {
57-
super.write(ctx, encodeHttp2HeadersFrame((Http2Header) msg), promise);
58-
} else if (msg instanceof Http2OutputMessage) {
59-
super.write(ctx, encodeHttp2DataFrame((Http2OutputMessage) msg), promise);
60-
} else {
61-
super.write(ctx, msg, promise);
84+
if (settingsFrameArrived) {
85+
if (msg instanceof Http2Header) {
86+
super.write(ctx, encodeHttp2HeadersFrame((Http2Header) msg), promise);
87+
} else if (msg instanceof Http2OutputMessage) {
88+
super.write(ctx, encodeHttp2DataFrame((Http2OutputMessage) msg), promise);
89+
} else {
90+
super.write(ctx, msg, promise);
91+
}
92+
return;
93+
}
94+
95+
if (LOGGER.isDebugEnabled()) {
96+
LOGGER.debug("Cache writing msg before client connection preface arrival: {}", msg);
97+
}
98+
cachedMsgList.add(new CachedMsg(ctx, msg, promise));
99+
100+
if (settingsFrameArrivalTimeoutFuture == null) {
101+
// close ctx and release resources if client connection preface does not arrive in time.
102+
settingsFrameArrivalTimeoutFuture = ctx.executor()
103+
.schedule(
104+
() -> {
105+
LOGGER.error(
106+
PROTOCOL_TIMEOUT_SERVER,
107+
"",
108+
"",
109+
"client connection preface does not arrive in time.");
110+
// send RST_STREAM instead of GO_AWAY by calling close method to avoid client hanging.
111+
ctx.close();
112+
nettyHttp2SettingsHandler.unsubscribeSettingsFrameArrival(this);
113+
cachedMsgList.clear();
114+
},
115+
SETTINGS_FRAME_ARRIVAL_TIMEOUT,
116+
TimeUnit.SECONDS);
62117
}
63118
}
64119

120+
public void notifySettingsFrameArrival() throws Exception {
121+
if (settingsFrameArrived) {
122+
return;
123+
}
124+
settingsFrameArrived = true;
125+
126+
if (settingsFrameArrivalTimeoutFuture != null) {
127+
settingsFrameArrivalTimeoutFuture.cancel(false);
128+
}
129+
130+
if (LOGGER.isDebugEnabled()) {
131+
LOGGER.debug("Begin cached channel msg writing when client connection preface arrived.");
132+
}
133+
134+
for (CachedMsg cached : cachedMsgList) {
135+
if (LOGGER.isDebugEnabled()) {
136+
LOGGER.debug("Cached channel msg writing, ctx: {} msg: {}", cached.ctx, cached.msg);
137+
}
138+
if (cached.msg instanceof Http2Header) {
139+
super.write(cached.ctx, encodeHttp2HeadersFrame(((Http2Header) cached.msg)), cached.promise);
140+
} else if (cached.msg instanceof Http2OutputMessage) {
141+
super.write(cached.ctx, encodeHttp2DataFrame(((Http2OutputMessage) cached.msg)), cached.promise);
142+
} else {
143+
super.write(cached.ctx, cached.msg, cached.promise);
144+
}
145+
}
146+
147+
if (LOGGER.isDebugEnabled()) {
148+
LOGGER.debug("End cached channel msg writing.");
149+
}
150+
151+
cachedMsgList.clear();
152+
}
153+
65154
private Http2Header onHttp2HeadersFrame(Http2HeadersFrame headersFrame) {
66155
return new Http2MetadataFrame(
67156
headersFrame.stream().id(), new DefaultHttpHeaders(headersFrame.headers()), headersFrame.isEndStream());
@@ -89,4 +178,16 @@ private Http2DataFrame encodeHttp2DataFrame(Http2OutputMessage outputMessage) {
89178
}
90179
throw new IllegalArgumentException("Http2OutputMessage body must be ByteBufOutputStream");
91180
}
181+
182+
private static class CachedMsg {
183+
private final ChannelHandlerContext ctx;
184+
private final Object msg;
185+
private final ChannelPromise promise;
186+
187+
public CachedMsg(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
188+
this.ctx = ctx;
189+
this.msg = msg;
190+
this.promise = promise;
191+
}
192+
}
92193
}

‎dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyHttp2SettingsHandler.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.dubbo.remoting.http12.netty4.h2;
18+
19+
import org.apache.dubbo.common.logger.Logger;
20+
import org.apache.dubbo.common.logger.LoggerFactory;
21+
22+
import java.util.HashSet;
23+
import java.util.Set;
24+
25+
import io.netty.channel.ChannelHandlerContext;
26+
import io.netty.channel.SimpleChannelInboundHandler;
27+
import io.netty.handler.codec.http2.Http2SettingsFrame;
28+
29+
/**
30+
* Add NettyHttp2SettingsHandler to pipeline because NettyHttp2FrameCodec could not receive Http2SettingsFrame.
31+
* Http2SettingsFrame does not belong to Http2StreamFrame or Http2GoAwayFrame that Http2MultiplexHandler
32+
* could process, NettyHttp2FrameCodec is wrapped in Http2MultiplexHandler as a child handler.
33+
*/
34+
public class NettyHttp2SettingsHandler extends SimpleChannelInboundHandler<Http2SettingsFrame> {
35+
36+
private static final Logger logger = LoggerFactory.getLogger(NettyHttp2SettingsHandler.class);
37+
38+
/**
39+
* Http2SettingsFrame arrival notification subscribers.
40+
*/
41+
private final Set<NettyHttp2FrameCodec> settingsFrameArrivalSubscribers = new HashSet<>();
42+
43+
private boolean settingsFrameArrived;
44+
45+
@Override
46+
protected void channelRead0(ChannelHandlerContext ctx, Http2SettingsFrame msg) throws Exception {
47+
if (logger.isDebugEnabled()) {
48+
logger.debug("Receive client Http2 Settings frame of "
49+
+ ctx.channel().localAddress() + " <- " + ctx.channel().remoteAddress());
50+
}
51+
settingsFrameArrived = true;
52+
53+
// Notify all subscribers that Http2SettingsFrame is arrived.
54+
for (NettyHttp2FrameCodec nettyHttp2FrameCodec : settingsFrameArrivalSubscribers) {
55+
nettyHttp2FrameCodec.notifySettingsFrameArrival();
56+
}
57+
settingsFrameArrivalSubscribers.clear();
58+
59+
ctx.pipeline().remove(this);
60+
}
61+
62+
/**
63+
* Save Http2SettingsFrame arrival notification subscriber if Http2SettingsFrame is not arrived.
64+
* @param nettyHttp2FrameCodec the netty HTTP2 frame codec that will be notified.
65+
* @return true: subscribe successful, false: Http2SettingsFrame arrived.
66+
*/
67+
public boolean subscribeSettingsFrameArrival(NettyHttp2FrameCodec nettyHttp2FrameCodec) {
68+
if (!settingsFrameArrived) {
69+
settingsFrameArrivalSubscribers.add(nettyHttp2FrameCodec);
70+
return true;
71+
}
72+
return false;
73+
}
74+
75+
public void unsubscribeSettingsFrameArrival(NettyHttp2FrameCodec nettyHttp2FrameCodec) {
76+
settingsFrameArrivalSubscribers.remove(nettyHttp2FrameCodec);
77+
}
78+
}

‎dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/http2/Http2ClientSettingsHandler.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,8 @@ public Http2ClientSettingsHandler(AtomicReference<Promise<Void>> connectionPrefa
3939
@Override
4040
protected void channelRead0(ChannelHandlerContext ctx, Http2SettingsFrame msg) throws Exception {
4141
if (logger.isDebugEnabled()) {
42-
logger.debug("Receive Http2 Settings frame of " + ctx.channel().localAddress() + " -> "
43-
+ ctx.channel().remoteAddress());
42+
logger.debug("Receive server Http2 Settings frame of "
43+
+ ctx.channel().localAddress() + " -> " + ctx.channel().remoteAddress());
4444
}
4545
// connectionPrefaceReceivedPromise will be set null after first used.
4646
Promise<Void> connectionPrefaceReceivedPromise = connectionPrefaceReceivedPromiseRef.get();
@@ -49,6 +49,7 @@ protected void channelRead0(ChannelHandlerContext ctx, Http2SettingsFrame msg) t
4949
} else {
5050
// Notify the connection preface is received when first inbound http2 settings frame is arrived.
5151
connectionPrefaceReceivedPromise.trySuccess(null);
52+
ctx.pipeline().remove(this);
5253
}
5354
}
5455
}

‎dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.apache.dubbo.remoting.http12.netty4.h1.NettyHttp1ConnectionHandler;
3232
import org.apache.dubbo.remoting.http12.netty4.h2.NettyHttp2FrameCodec;
3333
import org.apache.dubbo.remoting.http12.netty4.h2.NettyHttp2ProtocolSelectorHandler;
34+
import org.apache.dubbo.remoting.http12.netty4.h2.NettyHttp2SettingsHandler;
3435
import org.apache.dubbo.remoting.utils.UrlUtils;
3536
import org.apache.dubbo.remoting.websocket.netty4.WebSocketFrameCodec;
3637
import org.apache.dubbo.remoting.websocket.netty4.WebSocketProtocolSelectorHandler;
@@ -155,12 +156,14 @@ private void configurerHttp1Handlers(URL url, List<ChannelHandler> handlers) {
155156
sourceCodec,
156157
protocol -> {
157158
if (AsciiString.contentEquals(Http2CodecUtil.HTTP_UPGRADE_PROTOCOL_NAME, protocol)) {
159+
NettyHttp2SettingsHandler nettyHttp2SettingsHandler = new NettyHttp2SettingsHandler();
158160
return new Http2ServerUpgradeCodec(
159161
buildHttp2FrameCodec(tripleConfig),
162+
nettyHttp2SettingsHandler,
160163
new HttpWriteQueueHandler(),
161164
new FlushConsolidationHandler(64, true),
162165
new TripleServerConnectionHandler(),
163-
buildHttp2MultiplexHandler(url, tripleConfig),
166+
buildHttp2MultiplexHandler(nettyHttp2SettingsHandler, url, tripleConfig),
164167
new TripleTailHandler());
165168
} else if (AsciiString.contentEquals(HttpHeaderValues.WEBSOCKET, protocol)) {
166169
return new WebSocketServerUpgradeCodec(
@@ -191,24 +194,27 @@ private void configurerHttp1Handlers(URL url, List<ChannelHandler> handlers) {
191194
url, frameworkModel, tripleConfig, DefaultHttp11ServerTransportListenerFactory.INSTANCE)));
192195
}
193196

194-
private Http2MultiplexHandler buildHttp2MultiplexHandler(URL url, TripleConfig tripleConfig) {
197+
private Http2MultiplexHandler buildHttp2MultiplexHandler(
198+
NettyHttp2SettingsHandler nettyHttp2SettingsHandler, URL url, TripleConfig tripleConfig) {
195199
return new Http2MultiplexHandler(new ChannelInitializer<Http2StreamChannel>() {
196200
@Override
197201
protected void initChannel(Http2StreamChannel ch) {
198202
ChannelPipeline p = ch.pipeline();
199-
p.addLast(new NettyHttp2FrameCodec());
203+
p.addLast(new NettyHttp2FrameCodec(nettyHttp2SettingsHandler));
200204
p.addLast(new NettyHttp2ProtocolSelectorHandler(
201205
url, frameworkModel, tripleConfig, GenericHttp2ServerTransportListenerFactory.INSTANCE));
202206
}
203207
});
204208
}
205209

206210
private void configurerHttp2Handlers(URL url, List<ChannelHandler> handlers) {
211+
NettyHttp2SettingsHandler nettyHttp2SettingsHandler = new NettyHttp2SettingsHandler();
207212
TripleConfig tripleConfig = ConfigManager.getProtocolOrDefault(url).getTripleOrDefault();
208213
Http2FrameCodec codec = buildHttp2FrameCodec(tripleConfig);
209-
Http2MultiplexHandler handler = buildHttp2MultiplexHandler(url, tripleConfig);
214+
Http2MultiplexHandler handler = buildHttp2MultiplexHandler(nettyHttp2SettingsHandler, url, tripleConfig);
210215
handlers.add(new ChannelHandlerPretender(new HttpWriteQueueHandler()));
211216
handlers.add(new ChannelHandlerPretender(codec));
217+
handlers.add(new ChannelHandlerPretender(nettyHttp2SettingsHandler));
212218
handlers.add(new ChannelHandlerPretender(new FlushConsolidationHandler(64, true)));
213219
handlers.add(new ChannelHandlerPretender(new TripleServerConnectionHandler()));
214220
handlers.add(new ChannelHandlerPretender(handler));

0 commit comments

Comments
 (0)