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

Rewrite HttpObjectDecoder to make use of proper state machine

Browse Source 
Motivation:

HttpObjectDecoder extended ReplayDecoder which is slightly slower then ByteToMessageDecoder.

Modifications:

- Changed super class of HttpObjectDecoder from ReplayDecoder to ByteToMessageDecoder.
- Rewrote decode() method of HttpObjectDecoder to use proper state machine.
- Changed private methods HeaderParser.parse(ByteBuf), readHeaders(ByteBuf) and readTrailingHeaders(ByteBuf), skipControlCharacters(ByteBuf) to consider available bytes.
- Set HeaderParser and LineParser as static inner classes.
- Replaced not safe actualReadableBytes() with buffer.readableBytes().

Result:

Improved performance of HttpObjectDecoder by approximately 177%.
This commit is contained in:
Idel Pivnitskiy 2014-11-04 19:02:51 +03:00 committed by Norman Maurer
parent 229f83a7e6
commit b2ca2148f9
4 changed files with 90 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectDecoder.java
View File

@ -20,10 +20,9 @@ import io.netty.buffer.ByteBufProcessor;
import io.netty.buffer.Unpooled; import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DecoderResult; import io.netty.handler.codec.DecoderResult;
import io.netty.handler.codec.ReplayingDecoder;
import io.netty.handler.codec.TooLongFrameException; import io.netty.handler.codec.TooLongFrameException;
import io.netty.handler.codec.http.HttpObjectDecoder.State;
import io.netty.util.internal.AppendableCharSequence; import io.netty.util.internal.AppendableCharSequence;
import java.util.List; import java.util.List;
@ -99,7 +98,7 @@ import java.util.List;
* To implement the decoder of such a derived protocol, extend this class and * To implement the decoder of such a derived protocol, extend this class and
* implement all abstract methods properly. * implement all abstract methods properly.
*/ */
public abstract class HttpObjectDecoder extends ReplayingDecoder<State> { public abstract class HttpObjectDecoder extends ByteToMessageDecoder {
private static final String EMPTY_VALUE = ""; private static final String EMPTY_VALUE = "";
private final int maxChunkSize; private final int maxChunkSize;
@ -117,11 +116,13 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
private CharSequence name; private CharSequence name;
private CharSequence value; private CharSequence value;
private LastHttpContent trailer;
/** /**
* The internal state of {@link HttpObjectDecoder}. * The internal state of {@link HttpObjectDecoder}.
* <em>Internal use only</em>. * <em>Internal use only</em>.
*/ */
enum State { private enum State {
SKIP_CONTROL_CHARS, SKIP_CONTROL_CHARS,
READ_INITIAL, READ_INITIAL,
READ_HEADER, READ_HEADER,
@ -135,6 +136,8 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
UPGRADED UPGRADED
} }
private State currentState = State.SKIP_CONTROL_CHARS;
/** /**
* Creates a new instance with the default * Creates a new instance with the default
* {@code maxInitialLineLength (4096}}, {@code maxHeaderSize (8192)}, and * {@code maxInitialLineLength (4096}}, {@code maxHeaderSize (8192)}, and
@ -159,8 +162,6 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
int maxInitialLineLength, int maxHeaderSize, int maxChunkSize, int maxInitialLineLength, int maxHeaderSize, int maxChunkSize,
boolean chunkedSupported, boolean validateHeaders) { boolean chunkedSupported, boolean validateHeaders) {
super(State.SKIP_CONTROL_CHARS);
if (maxInitialLineLength <= 0) { if (maxInitialLineLength <= 0) {
throw new IllegalArgumentException( throw new IllegalArgumentException(
"maxInitialLineLength must be a positive integer: " + "maxInitialLineLength must be a positive integer: " +
@ -190,26 +191,27 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
resetNow(); resetNow();
} }
switch (state()) { switch (currentState) {
case SKIP_CONTROL_CHARS: { case SKIP_CONTROL_CHARS: {
try { if (!skipControlCharacters(buffer)) {
skipControlCharacters(buffer); return;
checkpoint(State.READ_INITIAL);
} finally {
checkpoint();
} }
// fall-through currentState = State.READ_INITIAL;
} }
case READ_INITIAL: try { case READ_INITIAL: try {
String[] initialLine = splitInitialLine(lineParser.parse(buffer)); AppendableCharSequence line = lineParser.parse(buffer);
if (line == null) {
return;
}
String[] initialLine = splitInitialLine(line);
if (initialLine.length < 3) { if (initialLine.length < 3) {
// Invalid initial line - ignore. // Invalid initial line - ignore.
checkpoint(State.SKIP_CONTROL_CHARS); currentState = State.SKIP_CONTROL_CHARS;
return; return;
} }
message = createMessage(initialLine); message = createMessage(initialLine);
checkpoint(State.READ_HEADER); currentState = State.READ_HEADER;
// fall-through // fall-through
} catch (Exception e) { } catch (Exception e) {
out.add(invalidMessage(e)); out.add(invalidMessage(e));
@ -217,7 +219,10 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
} }
case READ_HEADER: try { case READ_HEADER: try {
State nextState = readHeaders(buffer); State nextState = readHeaders(buffer);
checkpoint(nextState); if (nextState == null) {
return;
}
currentState = nextState;
switch (nextState) { switch (nextState) {
case SKIP_CONTROL_CHARS: case SKIP_CONTROL_CHARS:
// fast-path // fast-path
@ -261,7 +266,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
} }
case READ_VARIABLE_LENGTH_CONTENT: { case READ_VARIABLE_LENGTH_CONTENT: {
// Keep reading data as a chunk until the end of connection is reached. // Keep reading data as a chunk until the end of connection is reached.
int toRead = Math.min(actualReadableBytes(), maxChunkSize); int toRead = Math.min(buffer.readableBytes(), maxChunkSize);
if (toRead > 0) { if (toRead > 0) {
ByteBuf content = buffer.readSlice(toRead).retain(); ByteBuf content = buffer.readSlice(toRead).retain();
out.add(new DefaultHttpContent(content)); out.add(new DefaultHttpContent(content));
@ -269,7 +274,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
return; return;
} }
case READ_FIXED_LENGTH_CONTENT: { case READ_FIXED_LENGTH_CONTENT: {
int readLimit = actualReadableBytes(); int readLimit = buffer.readableBytes();
// Check if the buffer is readable first as we use the readable byte count // Check if the buffer is readable first as we use the readable byte count
// to create the HttpChunk. This is needed as otherwise we may end up with // to create the HttpChunk. This is needed as otherwise we may end up with
@ -303,13 +308,16 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
*/ */
case READ_CHUNK_SIZE: try { case READ_CHUNK_SIZE: try {
AppendableCharSequence line = lineParser.parse(buffer); AppendableCharSequence line = lineParser.parse(buffer);
if (line == null) {
return;
}
int chunkSize = getChunkSize(line.toString()); int chunkSize = getChunkSize(line.toString());
this.chunkSize = chunkSize; this.chunkSize = chunkSize;
if (chunkSize == 0) { if (chunkSize == 0) {
checkpoint(State.READ_CHUNK_FOOTER); currentState = State.READ_CHUNK_FOOTER;
return; return;
} }
checkpoint(State.READ_CHUNKED_CONTENT); currentState = State.READ_CHUNKED_CONTENT;
// fall-through // fall-through
} catch (Exception e) { } catch (Exception e) {
out.add(invalidChunk(e)); out.add(invalidChunk(e));
@ -318,7 +326,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
case READ_CHUNKED_CONTENT: { case READ_CHUNKED_CONTENT: {
assert chunkSize <= Integer.MAX_VALUE; assert chunkSize <= Integer.MAX_VALUE;
int toRead = Math.min((int) chunkSize, maxChunkSize); int toRead = Math.min((int) chunkSize, maxChunkSize);
toRead = Math.min(toRead, actualReadableBytes()); toRead = Math.min(toRead, buffer.readableBytes());
if (toRead == 0) { if (toRead == 0) {
return; return;
} }
@ -330,27 +338,27 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
if (chunkSize != 0) { if (chunkSize != 0) {
return; return;
} }
checkpoint(State.READ_CHUNK_DELIMITER); currentState = State.READ_CHUNK_DELIMITER;
// fall-through // fall-through
} }
case READ_CHUNK_DELIMITER: { case READ_CHUNK_DELIMITER: {
for (;;) { final int wIdx = buffer.writerIndex();
byte next = buffer.readByte(); int rIdx = buffer.readerIndex();
if (next == HttpConstants.CR) { while (wIdx > rIdx) {
if (buffer.readByte() == HttpConstants.LF) { byte next = buffer.getByte(rIdx++);
checkpoint(State.READ_CHUNK_SIZE); if (next == HttpConstants.LF) {
currentState = State.READ_CHUNK_SIZE;
break;
}
}
buffer.readerIndex(rIdx);
return; return;
} }
} else if (next == HttpConstants.LF) {
checkpoint(State.READ_CHUNK_SIZE);
return;
} else {
checkpoint();
}
}
}
case READ_CHUNK_FOOTER: try { case READ_CHUNK_FOOTER: try {
LastHttpContent trailer = readTrailingHeaders(buffer); LastHttpContent trailer = readTrailingHeaders(buffer);
if (trailer == null) {
return;
}
out.add(trailer); out.add(trailer);
resetNow(); resetNow();
return; return;
@ -360,17 +368,17 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
} }
case BAD_MESSAGE: { case BAD_MESSAGE: {
// Keep discarding until disconnection. // Keep discarding until disconnection.
buffer.skipBytes(actualReadableBytes()); buffer.skipBytes(buffer.readableBytes());
break; break;
} }
case UPGRADED: { case UPGRADED: {
int readableBytes = actualReadableBytes(); int readableBytes = buffer.readableBytes();
if (readableBytes > 0) { if (readableBytes > 0) {
// Keep on consuming as otherwise we may trigger an DecoderException, // Keep on consuming as otherwise we may trigger an DecoderException,
// other handler will replace this codec with the upgraded protocol codec to // other handler will replace this codec with the upgraded protocol codec to
// take the traffic over at some point then. // take the traffic over at some point then.
// See https://github.com/netty/netty/issues/2173 // See https://github.com/netty/netty/issues/2173
out.add(buffer.readBytes(actualReadableBytes())); out.add(buffer.readBytes(readableBytes));
} }
break; break;
} }
@ -384,7 +392,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
// Handle the last unfinished message. // Handle the last unfinished message.
if (message != null) { if (message != null) {
boolean chunked = HttpHeaderUtil.isTransferEncodingChunked(message); boolean chunked = HttpHeaderUtil.isTransferEncodingChunked(message);
if (state() == State.READ_VARIABLE_LENGTH_CONTENT && !in.isReadable() && !chunked) { if (currentState == State.READ_VARIABLE_LENGTH_CONTENT && !in.isReadable() && !chunked) {
// End of connection. // End of connection.
out.add(LastHttpContent.EMPTY_LAST_CONTENT); out.add(LastHttpContent.EMPTY_LAST_CONTENT);
reset(); reset();
@ -448,19 +456,20 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
contentLength = Long.MIN_VALUE; contentLength = Long.MIN_VALUE;
lineParser.reset(); lineParser.reset();
headerParser.reset(); headerParser.reset();
trailer = null;
if (!isDecodingRequest()) { if (!isDecodingRequest()) {
HttpResponse res = (HttpResponse) message; HttpResponse res = (HttpResponse) message;
if (res != null && res.status().code() == 101) { if (res != null && res.status().code() == 101) {
checkpoint(State.UPGRADED); currentState = State.UPGRADED;
return; return;
} }
} }
checkpoint(State.SKIP_CONTROL_CHARS); currentState = State.SKIP_CONTROL_CHARS;
} }
private HttpMessage invalidMessage(Exception cause) { private HttpMessage invalidMessage(Exception cause) {
checkpoint(State.BAD_MESSAGE); currentState = State.BAD_MESSAGE;
if (message != null) { if (message != null) {
message.setDecoderResult(DecoderResult.failure(cause)); message.setDecoderResult(DecoderResult.failure(cause));
} else { } else {
@ -474,22 +483,28 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
} }
private HttpContent invalidChunk(Exception cause) { private HttpContent invalidChunk(Exception cause) {
checkpoint(State.BAD_MESSAGE); currentState = State.BAD_MESSAGE;
HttpContent chunk = new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER); HttpContent chunk = new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER);
chunk.setDecoderResult(DecoderResult.failure(cause)); chunk.setDecoderResult(DecoderResult.failure(cause));
message = null; message = null;
trailer = null;
return chunk; return chunk;
} }
private static void skipControlCharacters(ByteBuf buffer) { private static boolean skipControlCharacters(ByteBuf buffer) {
for (;;) { boolean skiped = false;
char c = (char) buffer.readUnsignedByte(); final int wIdx = buffer.writerIndex();
if (!Character.isISOControl(c) && int rIdx = buffer.readerIndex();
!Character.isWhitespace(c)) { while (wIdx > rIdx) {
buffer.readerIndex(buffer.readerIndex() - 1); int c = buffer.getUnsignedByte(rIdx++);
if (!Character.isISOControl(c) && !Character.isWhitespace(c)) {
rIdx--;
skiped = true;
break; break;
} }
} }
buffer.readerIndex(rIdx);
return skiped;
} }
private State readHeaders(ByteBuf buffer) { private State readHeaders(ByteBuf buffer) {
@ -497,6 +512,9 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
final HttpHeaders headers = message.headers(); final HttpHeaders headers = message.headers();
AppendableCharSequence line = headerParser.parse(buffer); AppendableCharSequence line = headerParser.parse(buffer);
if (line == null) {
return null;
}
if (line.length() > 0) { if (line.length() > 0) {
do { do {
char firstChar = line.charAt(0); char firstChar = line.charAt(0);
@ -514,6 +532,9 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
} }
line = headerParser.parse(buffer); line = headerParser.parse(buffer);
if (line == null) {
return null;
}
} while (line.length() > 0); } while (line.length() > 0);
} }
@ -549,9 +570,15 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
private LastHttpContent readTrailingHeaders(ByteBuf buffer) { private LastHttpContent readTrailingHeaders(ByteBuf buffer) {
AppendableCharSequence line = headerParser.parse(buffer); AppendableCharSequence line = headerParser.parse(buffer);
if (line == null) {
return null;
}
CharSequence lastHeader = null; CharSequence lastHeader = null;
if (line.length() > 0) { if (line.length() > 0) {
LastHttpContent trailer = new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER, validateHeaders); LastHttpContent trailer = this.trailer;
if (trailer == null) {
trailer = this.trailer = new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER, validateHeaders);
}
do { do {
char firstChar = line.charAt(0); char firstChar = line.charAt(0);
if (lastHeader != null && (firstChar == ' ' || firstChar == '\t')) { if (lastHeader != null && (firstChar == ' ' || firstChar == '\t')) {
@ -582,8 +609,12 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
} }
line = headerParser.parse(buffer); line = headerParser.parse(buffer);
if (line == null) {
return null;
}
} while (line.length() > 0); } while (line.length() > 0);
this.trailer = null;
return trailer; return trailer;
} }
@ -693,7 +724,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
return result; return result;
} }
private class HeaderParser implements ByteBufProcessor { private static class HeaderParser implements ByteBufProcessor {
private final AppendableCharSequence seq; private final AppendableCharSequence seq;
private final int maxLength; private final int maxLength;
private int size; private int size;
@ -706,10 +737,10 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
public AppendableCharSequence parse(ByteBuf buffer) { public AppendableCharSequence parse(ByteBuf buffer) {
seq.reset(); seq.reset();
int i = buffer.forEachByte(this); int i = buffer.forEachByte(this);
if (i == -1) {
return null;
}
buffer.readerIndex(i + 1); buffer.readerIndex(i + 1);
// Call checkpoint to make sure the readerIndex is updated correctly
checkpoint();
return seq; return seq;
} }
@ -743,7 +774,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<State> {
} }
} }
private final class LineParser extends HeaderParser { private static final class LineParser extends HeaderParser {
LineParser(AppendableCharSequence seq, int maxLength) { LineParser(AppendableCharSequence seq, int maxLength) {
super(seq, maxLength); super(seq, maxLength);

2
codec-http/src/main/java/io/netty/handler/codec/http/HttpRequestDecoder.java
View File

@ -56,7 +56,7 @@ public class HttpRequestDecoder extends HttpObjectDecoder {
/** /**
* Creates a new instance with the default * Creates a new instance with the default
* {@code maxInitialLineLength (4096}}, {@code maxHeaderSize (8192)}, and * {@code maxInitialLineLength (4096)}, {@code maxHeaderSize (8192)}, and
* {@code maxChunkSize (8192)}. * {@code maxChunkSize (8192)}.
*/ */
public HttpRequestDecoder() { public HttpRequestDecoder() {

2
codec-http/src/main/java/io/netty/handler/codec/http/HttpResponseDecoder.java
View File

@ -87,7 +87,7 @@ public class HttpResponseDecoder extends HttpObjectDecoder {
/** /**
* Creates a new instance with the default * Creates a new instance with the default
* {@code maxInitialLineLength (4096}}, {@code maxHeaderSize (8192)}, and * {@code maxInitialLineLength (4096)}, {@code maxHeaderSize (8192)}, and
* {@code maxChunkSize (8192)}. * {@code maxChunkSize (8192)}.
*/ */
public HttpResponseDecoder() { public HttpResponseDecoder() {

2
codec-http/src/main/java/io/netty/handler/codec/rtsp/RtspObjectDecoder.java
View File

@ -52,7 +52,7 @@ public abstract class RtspObjectDecoder extends HttpObjectDecoder {
/** /**
* Creates a new instance with the default * Creates a new instance with the default
* {@code maxInitialLineLength (4096}}, {@code maxHeaderSize (8192)}, and * {@code maxInitialLineLength (4096)}, {@code maxHeaderSize (8192)}, and
* {@code maxContentLength (8192)}. * {@code maxContentLength (8192)}.
*/ */
protected RtspObjectDecoder() { protected RtspObjectDecoder() {