Merge pull request #280 from netty/cumulation_optimization
Make the cumulation usage more memory efficient
This commit is contained in:
commit
b4c00f0441
@ -17,7 +17,6 @@ package org.jboss.netty.handler.codec.frame;
|
||||
|
||||
import org.jboss.netty.buffer.ChannelBuffer;
|
||||
import org.jboss.netty.buffer.ChannelBufferFactory;
|
||||
import org.jboss.netty.buffer.ChannelBuffers;
|
||||
import org.jboss.netty.channel.Channel;
|
||||
import org.jboss.netty.channel.ChannelHandlerContext;
|
||||
|
||||
@ -78,8 +77,8 @@ public class FixedLengthFrameDecoder extends FrameDecoder {
|
||||
protected ChannelBuffer newCumulationBuffer(ChannelHandlerContext ctx, int minimumCapacity) {
|
||||
ChannelBufferFactory factory = ctx.getChannel().getConfig().getBufferFactory();
|
||||
if (allocateFullBuffer) {
|
||||
return ChannelBuffers.dynamicBuffer(
|
||||
factory.getDefaultOrder(), frameLength, ctx.getChannel().getConfig().getBufferFactory());
|
||||
return factory.getBuffer(frameLength);
|
||||
|
||||
}
|
||||
return super.newCumulationBuffer(ctx, minimumCapacity);
|
||||
}
|
||||
|
@ -30,7 +30,6 @@ import org.jboss.netty.channel.Channels;
|
||||
import org.jboss.netty.channel.ExceptionEvent;
|
||||
import org.jboss.netty.channel.MessageEvent;
|
||||
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
|
||||
import org.jboss.netty.handler.codec.replay.ReplayingDecoder;
|
||||
|
||||
/**
|
||||
* Decodes the received {@link ChannelBuffer}s into a meaningful frame object.
|
||||
@ -211,15 +210,47 @@ public abstract class FrameDecoder extends SimpleChannelUpstreamHandler {
|
||||
(this.cumulation = newCumulationBuffer(ctx, input.readableBytes())).writeBytes(input);
|
||||
}
|
||||
} else {
|
||||
ChannelBuffer cumulation = this.cumulation;
|
||||
assert cumulation.readable();
|
||||
if (cumulation.writableBytes() < input.readableBytes()) {
|
||||
cumulation.discardReadBytes();
|
||||
boolean fit = false;
|
||||
|
||||
int readable = input.readableBytes();
|
||||
int writable = cumulation.writableBytes();
|
||||
int w = writable - readable;
|
||||
if (w < 0) {
|
||||
int readerIndex = cumulation.readerIndex();
|
||||
if (w + readerIndex >= 0) {
|
||||
// the input will fit if we discard all read bytes, so do it
|
||||
cumulation.discardReadBytes();
|
||||
fit = true;
|
||||
}
|
||||
} else {
|
||||
|
||||
// ok the input fit into the cumulation buffer
|
||||
fit = true;
|
||||
}
|
||||
cumulation.writeBytes(input);
|
||||
callDecode(ctx, e.getChannel(), cumulation, e.getRemoteAddress());
|
||||
if (!cumulation.readable()) {
|
||||
|
||||
|
||||
ChannelBuffer buf;
|
||||
if (fit) {
|
||||
// the input fit in the cumulation buffer so copy it over
|
||||
buf = this.cumulation;
|
||||
buf.writeBytes(input);
|
||||
} else {
|
||||
// wrap the cumulation and input
|
||||
buf = ChannelBuffers.wrappedBuffer(cumulation, input);
|
||||
this.cumulation = buf;
|
||||
}
|
||||
|
||||
|
||||
callDecode(ctx, e.getChannel(), buf, e.getRemoteAddress());
|
||||
if (!buf.readable()) {
|
||||
// nothing readable left so reset the state
|
||||
this.cumulation = null;
|
||||
} else {
|
||||
// create a new buffer and copy the readable buffer into it
|
||||
this.cumulation = newCumulationBuffer(ctx, buf.readableBytes());
|
||||
this.cumulation.writeBytes(buf);
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -351,17 +382,14 @@ public abstract class FrameDecoder extends SimpleChannelUpstreamHandler {
|
||||
|
||||
/**
|
||||
* Create a new {@link ChannelBuffer} which is used for the cumulation.
|
||||
* Be aware that this MUST be a dynamic buffer. Sub-classes may override
|
||||
* this to provide a dynamic {@link ChannelBuffer} which has some
|
||||
* pre-allocated size that better fit their need.
|
||||
*
|
||||
* Sub-classes may override this.
|
||||
*
|
||||
* @param ctx {@link ChannelHandlerContext} for this handler
|
||||
* @return buffer the {@link ChannelBuffer} which is used for cumulation
|
||||
*/
|
||||
protected ChannelBuffer newCumulationBuffer(
|
||||
ChannelHandlerContext ctx, int minimumCapacity) {
|
||||
ChannelBufferFactory factory = ctx.getChannel().getConfig().getBufferFactory();
|
||||
return ChannelBuffers.dynamicBuffer(
|
||||
factory.getDefaultOrder(), Math.max(minimumCapacity, 256), factory);
|
||||
return factory.getBuffer(Math.max(minimumCapacity, 256));
|
||||
}
|
||||
}
|
||||
|
@ -474,16 +474,49 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
|
||||
replayable = ReplayingDecoderBuffer.EMPTY_BUFFER;
|
||||
}
|
||||
} else {
|
||||
ChannelBuffer cumulation = this.cumulation;
|
||||
assert cumulation.readable();
|
||||
if (cumulation.writableBytes() < input.readableBytes()) {
|
||||
cumulation.discardReadBytes();
|
||||
boolean fit = false;
|
||||
|
||||
int readable = input.readableBytes();
|
||||
int writable = cumulation.writableBytes();
|
||||
int w = writable - readable;
|
||||
if (w < 0) {
|
||||
int readerIndex = cumulation.readerIndex();
|
||||
if (w + readerIndex >= 0) {
|
||||
// the input will fit if we discard all read bytes, so do it
|
||||
cumulation.discardReadBytes();
|
||||
fit = true;
|
||||
}
|
||||
} else {
|
||||
|
||||
// ok the input fit into the cumulation buffer
|
||||
fit = true;
|
||||
}
|
||||
cumulation.writeBytes(input);
|
||||
callDecode(ctx, e.getChannel(), cumulation, replayable, e.getRemoteAddress());
|
||||
if (!cumulation.readable()) {
|
||||
|
||||
ChannelBuffer buf;
|
||||
if (fit) {
|
||||
// the input fit in the cumulation buffer so copy it over
|
||||
buf = this.cumulation;
|
||||
buf.writeBytes(input);
|
||||
} else {
|
||||
// wrap the cumulation and input
|
||||
buf = ChannelBuffers.wrappedBuffer(cumulation, input);
|
||||
this.cumulation = buf;
|
||||
replayable = new ReplayingDecoderBuffer(cumulation);
|
||||
}
|
||||
|
||||
|
||||
callDecode(ctx, e.getChannel(), buf, replayable, e.getRemoteAddress());
|
||||
if (!buf.readable()) {
|
||||
// nothing readable left so reset the state
|
||||
this.cumulation = null;
|
||||
replayable = ReplayingDecoderBuffer.EMPTY_BUFFER;
|
||||
} else {
|
||||
// create a new buffer and copy the readable buffer into it
|
||||
this.cumulation = newCumulationBuffer(ctx, buf.readableBytes());
|
||||
this.cumulation.writeBytes(buf);
|
||||
replayable = new ReplayingDecoderBuffer(this.cumulation);
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -606,9 +639,7 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
|
||||
|
||||
/**
|
||||
* Create a new {@link ChannelBuffer} which is used for the cumulation.
|
||||
* Be aware that this MUST be a dynamic buffer. Sub-classes may override
|
||||
* this to provide a dynamic {@link ChannelBuffer} which has some
|
||||
* pre-allocated size that better fit their need.
|
||||
* Sub-classes may override this.
|
||||
*
|
||||
* @param ctx {@link ChannelHandlerContext} for this handler
|
||||
* @return buffer the {@link ChannelBuffer} which is used for cumulation
|
||||
@ -616,7 +647,6 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
|
||||
protected ChannelBuffer newCumulationBuffer(
|
||||
ChannelHandlerContext ctx, int minimumCapacity) {
|
||||
ChannelBufferFactory factory = ctx.getChannel().getConfig().getBufferFactory();
|
||||
return ChannelBuffers.dynamicBuffer(
|
||||
factory.getDefaultOrder(), Math.max(minimumCapacity, 256), factory);
|
||||
return factory.getBuffer(Math.max(minimumCapacity, 256));
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user