andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Reduce memory copies in HttpContentDecoder and so also the risk of memory leaks

Browse Source
This commit is contained in:
Norman Maurer 2013-07-14 23:12:49 +02:00
parent ecb215c12f
commit 0393ffbfb2
1 changed files with 41 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java
View File

@ -16,8 +16,6 @@
package io.netty.handler.codec.http; package io.netty.handler.codec.http;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.embedded.EmbeddedChannel; import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.MessageToMessageDecoder; import io.netty.handler.codec.MessageToMessageDecoder;
@ -108,15 +106,23 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
headers.set(HttpHeaders.Names.CONTENT_ENCODING, targetContentEncoding); headers.set(HttpHeaders.Names.CONTENT_ENCODING, targetContentEncoding);
} }
Object[] decoded = decodeContent(message, c); out.add(message);
decodeContent(c, out);
// Replace the content. // Replace the content length.
if (headers.contains(HttpHeaders.Names.CONTENT_LENGTH)) { if (headers.contains(HttpHeaders.Names.CONTENT_LENGTH)) {
int contentLength = 0;
int size = out.size();
for (int i = 0; i < size; i++) {
Object o = out.get(i);
if (o instanceof HttpContent) {
contentLength += ((HttpContent) o).content().readableBytes();
}
}
headers.set( headers.set(
HttpHeaders.Names.CONTENT_LENGTH, HttpHeaders.Names.CONTENT_LENGTH,
Integer.toString(((ByteBufHolder) decoded[1]).content().readableBytes())); Integer.toString(contentLength));
} }
out.add(decoded);
return; return;
} }
@ -129,7 +135,7 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
} }
if (decoder != null) { if (decoder != null) {
out.add(decodeContent(null, c)); decodeContent(c, out);
} else { } else {
if (c instanceof LastHttpContent) { if (c instanceof LastHttpContent) {
decodeStarted = false; decodeStarted = false;
@ -139,36 +145,17 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
} }
} }
private Object[] decodeContent(HttpMessage header, HttpContent c) { private void decodeContent(HttpContent c, List<Object> out) {
ByteBuf newContent = Unpooled.buffer();
ByteBuf content = c.content(); ByteBuf content = c.content();
decode(content, newContent);
decode(content, out);
if (c instanceof LastHttpContent) { if (c instanceof LastHttpContent) {
ByteBuf lastProduct = Unpooled.buffer(); finishDecode(out);
finishDecode(lastProduct);
// Generate an additional chunk if the decoder produced // Generate an additional chunk if the decoder produced
// the last product on closure, // the last product on closure,
if (lastProduct.isReadable()) { out.add(LastHttpContent.EMPTY_LAST_CONTENT);
if (header == null) {
return new Object[] { new DefaultHttpContent(newContent), new DefaultLastHttpContent(lastProduct)};
} else {
return new Object[] { header, new DefaultHttpContent(newContent),
new DefaultLastHttpContent(lastProduct)};
}
} else {
if (header == null) {
return new Object[] { new DefaultLastHttpContent(newContent) };
} else {
return new Object[] { header, new DefaultLastHttpContent(newContent) };
}
}
}
if (header == null) {
return new Object[] { new DefaultHttpContent(newContent) };
} else {
return new Object[] { header, new DefaultHttpContent(newContent) };
} }
} }
@ -210,20 +197,28 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
private void cleanup() { private void cleanup() {
if (decoder != null) { if (decoder != null) {
// Clean-up the previous decoder if not cleaned up correctly. // Clean-up the previous encoder if not cleaned up correctly.
ByteBuf buf = Unpooled.buffer(); if (decoder.finish()) {
finishDecode(buf); for (;;) {
ByteBuf buf = (ByteBuf) decoder.readOutbound();
if (buf == null) {
break;
}
// Release the buffer
buf.release(); buf.release();
} }
} }
decoder = null;
}
}
private void decode(ByteBuf in, ByteBuf out) { private void decode(ByteBuf in, List<Object> out) {
// call retain as it will be release after is written // call retain here as it will call release after its written to the channel
decoder.writeInbound(in.retain()); decoder.writeOutbound(in.retain());
fetchDecoderOutput(out); fetchDecoderOutput(out);
} }
private void finishDecode(ByteBuf out) { private void finishDecode(List<Object> out) {
if (decoder.finish()) { if (decoder.finish()) {
fetchDecoderOutput(out); fetchDecoderOutput(out);
} }
@ -231,14 +226,17 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
decoder = null; decoder = null;
} }
private void fetchDecoderOutput(ByteBuf out) { private void fetchDecoderOutput(List<Object> out) {
for (;;) { for (;;) {
ByteBuf buf = (ByteBuf) decoder.readInbound(); ByteBuf buf = (ByteBuf) decoder.readOutbound();
if (buf == null) { if (buf == null) {
break; break;
} }
out.writeBytes(buf); if (!buf.isReadable()) {
buf.release(); buf.release();
continue;
}
out.add(new DefaultHttpContent(buf));
} }
} }
} }