Skip to content

Commit acdbb07

Browse files
authored
fix TcpDecoder memory leak (#54)
* fix TcpDecoder0 memory leak * update
1 parent e90e472 commit acdbb07

File tree

3 files changed

+91
-2
lines changed

3 files changed

+91
-2
lines changed

trpc-transport/trpc-transport-netty/src/main/java/com/tencent/trpc/transport/netty/NettyChannel.java

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,12 @@ public NettyChannel(io.netty.channel.Channel channel, ProtocolConfig config) {
3535
this.config = config;
3636
if (channel != null) {
3737
// can't get the remote address while using udp, so the remoteAddress is null
38-
this.remoteAddress = ((InetSocketAddress) channel.remoteAddress());
39-
this.localAddress = (InetSocketAddress) channel.localAddress();
38+
if (channel.remoteAddress() instanceof InetSocketAddress) {
39+
this.remoteAddress = ((InetSocketAddress) channel.remoteAddress());
40+
}
41+
if (channel.localAddress() instanceof InetSocketAddress) {
42+
this.localAddress = (InetSocketAddress) channel.localAddress();
43+
}
4044
}
4145
// listen for the close event
4246
if (channel != null && channel.closeFuture() != null) {

trpc-transport/trpc-transport-netty/src/main/java/com/tencent/trpc/transport/netty/NettyCodecAdapter.java

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,9 @@ private void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out)
170170
}
171171
}
172172
} while (message.isReadable());
173+
} catch (Exception e) {
174+
message.skipBytes(message.readableBytes());
175+
throw new TransportException("tcp|decode failure", e);
173176
} finally {
174177
NettyChannelManager.removeChannelIfDisconnected(ctx.channel());
175178
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
package com.tencent.trpc.transport.netty;
2+
3+
import com.tencent.trpc.core.common.config.ProtocolConfig;
4+
import com.tencent.trpc.core.exception.ErrorCode;
5+
import com.tencent.trpc.core.exception.TRpcException;
6+
import com.tencent.trpc.core.exception.TransportException;
7+
import com.tencent.trpc.core.transport.codec.Codec;
8+
import io.netty.buffer.AbstractByteBufAllocator;
9+
import io.netty.buffer.ByteBuf;
10+
import io.netty.channel.ChannelHandler;
11+
import io.netty.channel.embedded.EmbeddedChannel;
12+
import io.netty.handler.codec.DecoderException;
13+
import org.junit.Assert;
14+
import org.junit.Test;
15+
16+
import java.nio.charset.StandardCharsets;
17+
18+
import static org.mockito.Matchers.any;
19+
import static org.mockito.Mockito.doThrow;
20+
import static org.mockito.Mockito.mock;
21+
22+
23+
public class NettyCodecAdapterTest {
24+
25+
@Test
26+
public void testTcpDecodeIllegalPacket1() {
27+
Codec codec = mock(Codec.class);
28+
doThrow(TRpcException.newFrameException(ErrorCode.TRPC_CLIENT_DECODE_ERR, "the request protocol is not trpc"))
29+
.when(codec).decode(any(), any());
30+
31+
32+
ProtocolConfig protocolConfig = new ProtocolConfig();
33+
// set batchDecoder true
34+
protocolConfig.setBatchDecoder(true);
35+
NettyCodecAdapter nettyCodecAdapter = NettyCodecAdapter.createTcpCodecAdapter(codec, protocolConfig);
36+
37+
ChannelHandler decoder = nettyCodecAdapter.getDecoder();
38+
EmbeddedChannel embeddedChannel = new EmbeddedChannel();
39+
embeddedChannel.pipeline().addLast(decoder);
40+
41+
ByteBuf byteBuf = AbstractByteBufAllocator.DEFAULT.heapBuffer();
42+
byteBuf.writeBytes("testTcpDecodeIllegalPacket1".getBytes(StandardCharsets.UTF_8));
43+
44+
// write illegal packet
45+
EmbeddedChannel tmpEmbeddedChannel = embeddedChannel;
46+
DecoderException decoderException = Assert.assertThrows(DecoderException.class, () -> {
47+
tmpEmbeddedChannel.writeInbound(byteBuf);
48+
});
49+
50+
Assert.assertTrue(decoderException.getCause() instanceof TransportException);
51+
Assert.assertEquals(byteBuf.refCnt(), 0);
52+
}
53+
54+
@Test
55+
public void testTcpDecodeIllegalPacket2() {
56+
Codec codec = mock(Codec.class);
57+
doThrow(TRpcException.newFrameException(ErrorCode.TRPC_CLIENT_DECODE_ERR, "the request protocol is not trpc"))
58+
.when(codec).decode(any(), any());
59+
60+
61+
ProtocolConfig protocolConfig = new ProtocolConfig();
62+
// set batchDecoder false
63+
protocolConfig.setBatchDecoder(false);
64+
NettyCodecAdapter nettyCodecAdapter = NettyCodecAdapter.createTcpCodecAdapter(codec, protocolConfig);
65+
66+
ChannelHandler decoder = nettyCodecAdapter.getDecoder();
67+
EmbeddedChannel embeddedChannel = new EmbeddedChannel();
68+
embeddedChannel.pipeline().addLast(decoder);
69+
70+
ByteBuf byteBuf = AbstractByteBufAllocator.DEFAULT.heapBuffer();
71+
byteBuf.writeBytes("testTcpDecodeIllegalPacket1".getBytes(StandardCharsets.UTF_8));
72+
73+
// write illegal packet
74+
EmbeddedChannel tmpEmbeddedChannel = embeddedChannel;
75+
DecoderException decoderException = Assert.assertThrows(DecoderException.class, () -> {
76+
tmpEmbeddedChannel.writeInbound(byteBuf);
77+
});
78+
79+
Assert.assertTrue(decoderException.getCause() instanceof TransportException);
80+
Assert.assertEquals(byteBuf.refCnt(), 0);
81+
}
82+
}

0 commit comments

Comments
 (0)