Skip to content

Commit ac4dacd

Browse files
normanmaurernormanmaurer
committed
Minimize byte copies by using a CompositeByteBuf to concat the chunks. See netty#413 
1 parent d0e8352 commit ac4dacd

File tree

2 files changed

+205
-2
lines changed

2 files changed

+205
-2
lines changed

‎codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java

+65-2
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import static io.netty.handler.codec.http.HttpHeaders.*;
1919
import io.netty.buffer.ByteBuf;
20+
import io.netty.buffer.CompositeByteBuf;
2021
import io.netty.buffer.Unpooled;
2122
import io.netty.channel.ChannelHandler;
2223
import io.netty.channel.ChannelHandlerContext;
@@ -47,13 +48,16 @@
4748
* @apiviz.has io.netty.handler.codec.http.HttpChunk oneway - - filters out
4849
*/
4950
public class HttpChunkAggregator extends MessageToMessageDecoder<Object, HttpMessage> {
50-
51+
public static final int DEFAULT_MAX_COMPOSITEBUFFER_COMPONENTS = 1024;
5152
private static final ByteBuf CONTINUE = Unpooled.copiedBuffer(
5253
"HTTP/1.1 100 Continue\r\n\r\n", CharsetUtil.US_ASCII);
5354

5455
private final int maxContentLength;
5556
private HttpMessage currentMessage;
5657

58+
private int maxCumulationBufferComponents = DEFAULT_MAX_COMPOSITEBUFFER_COMPONENTS;
59+
private ChannelHandlerContext ctx;
60+
5761
/**
5862
* Creates a new instance.
5963
*
@@ -71,6 +75,38 @@ public HttpChunkAggregator(int maxContentLength) {
7175
this.maxContentLength = maxContentLength;
7276
}
7377

78+
/**
79+
* Returns the maximum number of components in the cumulation buffer. If the number of
80+
* the components in the cumulation buffer exceeds this value, the components of the
81+
* cumulation buffer are consolidated into a single component, involving memory copies.
82+
* The default value of this property is {@link #DEFAULT_MAX_COMPOSITEBUFFER_COMPONENTS}.
83+
*/
84+
public final int getMaxCumulationBufferComponents() {
85+
return maxCumulationBufferComponents;
86+
}
87+
88+
/**
89+
* Sets the maximum number of components in the cumulation buffer. If the number of
90+
* the components in the cumulation buffer exceeds this value, the components of the
91+
* cumulation buffer are consolidated into a single component, involving memory copies.
92+
* The default value of this property is {@link #DEFAULT_MAX_COMPOSITEBUFFER_COMPONENTS}
93+
* and its minimum allowed value is {@code 2}.
94+
*/
95+
public final void setMaxCumulationBufferComponents(int maxCumulationBufferComponents) {
96+
if (maxCumulationBufferComponents < 2) {
97+
throw new IllegalArgumentException(
98+
"maxCumulationBufferComponents: " + maxCumulationBufferComponents +
99+
" (expected: >= 2)");
100+
}
101+
102+
if (ctx == null) {
103+
this.maxCumulationBufferComponents = maxCumulationBufferComponents;
104+
} else {
105+
throw new IllegalStateException(
106+
"decoder properties cannot be changed once the decoder is added to a pipeline.");
107+
}
108+
}
109+
74110
@Override
75111
public boolean isDecodable(Object msg) throws Exception {
76112
return msg instanceof HttpMessage || msg instanceof HttpChunk;
@@ -131,7 +167,9 @@ public HttpMessage decode(ChannelHandlerContext ctx, Object msg) throws Exceptio
131167
" bytes.");
132168
}
133169

134-
content.writeBytes(chunk.getContent());
170+
// Append the content of the chunk
171+
appendToCumulation(chunk.getContent());
172+
135173
if (chunk.isLast()) {
136174
this.currentMessage = null;
137175

@@ -159,4 +197,29 @@ public HttpMessage decode(ChannelHandlerContext ctx, Object msg) throws Exceptio
159197
HttpChunk.class.getSimpleName() + " are accepted: " + msg.getClass().getName());
160198
}
161199
}
200+
201+
protected void appendToCumulation(ByteBuf input) {
202+
ByteBuf cumulation = this.currentMessage.getContent();
203+
if (cumulation instanceof CompositeByteBuf) {
204+
// Make sure the resulting cumulation buffer has no more than 4 components.
205+
CompositeByteBuf composite = (CompositeByteBuf) cumulation;
206+
if (composite.numComponents() >= maxCumulationBufferComponents) {
207+
currentMessage.setContent(Unpooled.wrappedBuffer(composite.copy(), input));
208+
} else {
209+
List<ByteBuf> decomposed = composite.decompose(0, composite.readableBytes());
210+
ByteBuf[] buffers = decomposed.toArray(new ByteBuf[decomposed.size() + 1]);
211+
buffers[buffers.length - 1] = input;
212+
213+
currentMessage.setContent(Unpooled.wrappedBuffer(buffers));
214+
}
215+
} else {
216+
currentMessage.setContent(Unpooled.wrappedBuffer(cumulation, input));
217+
}
218+
219+
}
220+
221+
public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
222+
this.ctx = ctx;
223+
}
224+
162225
}

‎codec-http/src/test/java/io/netty/handler/codec/http/HttpChunkAggregatorTest.java

+140
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
/*
2+
* Copyright 2012 The Netty Project
3+
*
4+
* The Netty Project licenses this file to you under the Apache License,
5+
* version 2.0 (the "License"); you may not use this file except in compliance
6+
* with the License. You may obtain a copy of the License at:
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations
14+
* under the License.
15+
*/
16+
package io.netty.handler.codec.http;
17+
18+
import static org.junit.Assert.*;
19+
20+
import java.util.List;
21+
22+
import org.easymock.EasyMock;
23+
import io.netty.buffer.ByteBuf;
24+
import io.netty.buffer.Unpooled;
25+
import io.netty.buffer.CompositeByteBuf;
26+
import io.netty.channel.ChannelHandlerContext;
27+
import io.netty.channel.embedded.EmbeddedMessageChannel;
28+
import io.netty.handler.codec.TooLongFrameException;
29+
30+
import io.netty.util.CharsetUtil;
31+
import org.junit.Test;
32+
33+
public class HttpChunkAggregatorTest {
34+
35+
@Test
36+
public void testAggregate() {
37+
HttpChunkAggregator aggr = new HttpChunkAggregator(1024 * 1024);
38+
EmbeddedMessageChannel embedder = new EmbeddedMessageChannel(aggr);
39+
40+
HttpMessage message = new DefaultHttpMessage(HttpVersion.HTTP_1_1);
41+
HttpHeaders.setHeader(message, "X-Test", true);
42+
message.setChunked(true);
43+
HttpChunk chunk1 = new DefaultHttpChunk(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII));
44+
HttpChunk chunk2 = new DefaultHttpChunk(Unpooled.copiedBuffer("test2", CharsetUtil.US_ASCII));
45+
HttpChunk chunk3 = new DefaultHttpChunk(Unpooled.EMPTY_BUFFER);
46+
assertFalse(embedder.writeInbound(message));
47+
assertFalse(embedder.writeInbound(chunk1));
48+
assertFalse(embedder.writeInbound(chunk2));
49+
50+
// this should trigger a messageReceived event so return true
51+
assertTrue(embedder.writeInbound(chunk3));
52+
assertTrue(embedder.finish());
53+
HttpMessage aggratedMessage = (HttpMessage) embedder.readInbound();
54+
assertNotNull(aggratedMessage);
55+
56+
assertEquals(chunk1.getContent().readableBytes() + chunk2.getContent().readableBytes(), HttpHeaders.getContentLength(aggratedMessage));
57+
assertEquals(aggratedMessage.getHeader("X-Test"), Boolean.TRUE.toString());
58+
checkContentBuffer(aggratedMessage);
59+
assertNull(embedder.readInbound());
60+
61+
}
62+
63+
private void checkContentBuffer(HttpMessage aggregatedMessage) {
64+
CompositeByteBuf buffer = (CompositeByteBuf) aggregatedMessage.getContent();
65+
assertEquals(2, buffer.numComponents());
66+
List<ByteBuf> buffers = buffer.decompose(0, buffer.capacity());
67+
assertEquals(2, buffers.size());
68+
for (ByteBuf buf: buffers) {
69+
// This should be false as we decompose the buffer before to not have deep hierarchy
70+
assertFalse(buf instanceof CompositeByteBuf);
71+
}
72+
}
73+
74+
@Test
75+
public void testAggregateWithTrailer() {
76+
HttpChunkAggregator aggr = new HttpChunkAggregator(1024 * 1024);
77+
EmbeddedMessageChannel embedder = new EmbeddedMessageChannel(aggr);
78+
HttpMessage message = new DefaultHttpMessage(HttpVersion.HTTP_1_1);
79+
HttpHeaders.setHeader(message, "X-Test", true);
80+
message.setChunked(true);
81+
HttpChunk chunk1 = new DefaultHttpChunk(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII));
82+
HttpChunk chunk2 = new DefaultHttpChunk(Unpooled.copiedBuffer("test2", CharsetUtil.US_ASCII));
83+
HttpChunkTrailer trailer = new DefaultHttpChunkTrailer();
84+
trailer.setHeader("X-Trailer", true);
85+
86+
assertFalse(embedder.writeInbound(message));
87+
assertFalse(embedder.writeInbound(chunk1));
88+
assertFalse(embedder.writeInbound(chunk2));
89+
90+
// this should trigger a messageReceived event so return true
91+
assertTrue(embedder.writeInbound(trailer));
92+
assertTrue(embedder.finish());
93+
HttpMessage aggratedMessage = (HttpMessage) embedder.readInbound();
94+
assertNotNull(aggratedMessage);
95+
96+
assertEquals(chunk1.getContent().readableBytes() + chunk2.getContent().readableBytes(), HttpHeaders.getContentLength(aggratedMessage));
97+
assertEquals(aggratedMessage.getHeader("X-Test"), Boolean.TRUE.toString());
98+
assertEquals(aggratedMessage.getHeader("X-Trailer"), Boolean.TRUE.toString());
99+
checkContentBuffer(aggratedMessage);
100+
101+
assertNull(embedder.readInbound());
102+
103+
}
104+
105+
106+
@Test(expected = TooLongFrameException.class)
107+
public void testTooLongFrameException() {
108+
HttpChunkAggregator aggr = new HttpChunkAggregator(4);
109+
EmbeddedMessageChannel embedder = new EmbeddedMessageChannel(aggr);
110+
HttpMessage message = new DefaultHttpMessage(HttpVersion.HTTP_1_1);
111+
message.setChunked(true);
112+
HttpChunk chunk1 = new DefaultHttpChunk(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII));
113+
HttpChunk chunk2 = new DefaultHttpChunk(Unpooled.copiedBuffer("test2", CharsetUtil.US_ASCII));
114+
assertFalse(embedder.writeInbound(message));
115+
assertFalse(embedder.writeInbound(chunk1));
116+
embedder.writeInbound(chunk2);
117+
fail();
118+
119+
}
120+
121+
@Test(expected = IllegalArgumentException.class)
122+
public void testInvalidConstructorUsage() {
123+
new HttpChunkAggregator(0);
124+
}
125+
126+
@Test(expected = IllegalArgumentException.class)
127+
public void testInvalidMaxCumulationBufferComponents() {
128+
HttpChunkAggregator aggr= new HttpChunkAggregator(Integer.MAX_VALUE);
129+
aggr.setMaxCumulationBufferComponents(1);
130+
}
131+
132+
@Test(expected = IllegalStateException.class)
133+
public void testSetMaxCumulationBufferComponentsAfterInit() throws Exception {
134+
HttpChunkAggregator aggr = new HttpChunkAggregator(Integer.MAX_VALUE);
135+
ChannelHandlerContext ctx = EasyMock.createMock(ChannelHandlerContext.class);
136+
EasyMock.replay(ctx);
137+
aggr.beforeAdd(ctx);
138+
aggr.setMaxCumulationBufferComponents(10);
139+
}
140+
}

0 commit comments

Comments
 (0)