Correctly fix problem in ByteToMessageDecoder and ReplayingDecoder which could let to have a released buffer passed to the decode methods.
This fixes #1664 and revert also the original commit which was meant to fix it 3b1881b523
. The problem with the original commit was that it could delay handlerRemove(..) calls and so mess up the order or forward bytes to late.
This commit is contained in:
parent
e3410680de
commit
a06295fe0a
@ -229,6 +229,15 @@ public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter
|
|||||||
int outSize = out.size();
|
int outSize = out.size();
|
||||||
int oldInputLength = in.readableBytes();
|
int oldInputLength = in.readableBytes();
|
||||||
decode(ctx, in, out);
|
decode(ctx, in, out);
|
||||||
|
|
||||||
|
// Check if this handler was removed before try to continue the loop.
|
||||||
|
// If it was removed it is not safe to continue to operate on the buffer
|
||||||
|
//
|
||||||
|
// See https://github.com/netty/netty/issues/1664
|
||||||
|
if (ctx.isRemoved()) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
if (outSize == out.size()) {
|
if (outSize == out.size()) {
|
||||||
if (oldInputLength == in.readableBytes()) {
|
if (oldInputLength == in.readableBytes()) {
|
||||||
break;
|
break;
|
||||||
|
@ -358,6 +358,15 @@ public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {
|
|||||||
int oldInputLength = in.readableBytes();
|
int oldInputLength = in.readableBytes();
|
||||||
try {
|
try {
|
||||||
decode(ctx, replayable, out);
|
decode(ctx, replayable, out);
|
||||||
|
|
||||||
|
// Check if this handler was removed before try to continue the loop.
|
||||||
|
// If it was removed it is not safe to continue to operate on the buffer
|
||||||
|
//
|
||||||
|
// See https://github.com/netty/netty/issues/1664
|
||||||
|
if (ctx.isRemoved()) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
if (outSize == out.size()) {
|
if (outSize == out.size()) {
|
||||||
if (oldInputLength == in.readableBytes() && oldState == state) {
|
if (oldInputLength == in.readableBytes() && oldState == state) {
|
||||||
throw new DecoderException(
|
throw new DecoderException(
|
||||||
@ -371,6 +380,15 @@ public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {
|
|||||||
}
|
}
|
||||||
} catch (Signal replay) {
|
} catch (Signal replay) {
|
||||||
replay.expect(REPLAY);
|
replay.expect(REPLAY);
|
||||||
|
|
||||||
|
// Check if this handler was removed before try to continue the loop.
|
||||||
|
// If it was removed it is not safe to continue to operate on the buffer
|
||||||
|
//
|
||||||
|
// See https://github.com/netty/netty/issues/1664
|
||||||
|
if (ctx.isRemoved()) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
// Return to the checkpoint (or oldPosition) and retry.
|
// Return to the checkpoint (or oldPosition) and retry.
|
||||||
int checkpoint = this.checkpoint;
|
int checkpoint = this.checkpoint;
|
||||||
if (checkpoint >= 0) {
|
if (checkpoint >= 0) {
|
||||||
|
@ -34,7 +34,6 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
private final DefaultChannelPipeline pipeline;
|
private final DefaultChannelPipeline pipeline;
|
||||||
private final String name;
|
private final String name;
|
||||||
private final ChannelHandler handler;
|
private final ChannelHandler handler;
|
||||||
private int callDepth;
|
|
||||||
private boolean removed;
|
private boolean removed;
|
||||||
|
|
||||||
// Will be set to null if no child executor should be used, otherwise it will be set to the
|
// Will be set to null if no child executor should be used, otherwise it will be set to the
|
||||||
@ -78,10 +77,6 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
boolean isRunning() {
|
|
||||||
return callDepth != 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
/** Invocation initiated by {@link DefaultChannelPipeline#teardownAll()}}. */
|
/** Invocation initiated by {@link DefaultChannelPipeline#teardownAll()}}. */
|
||||||
void teardown() {
|
void teardown() {
|
||||||
EventExecutor executor = executor();
|
EventExecutor executor = executor();
|
||||||
@ -160,12 +155,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeChannelRegistered() {
|
private void invokeChannelRegistered() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).channelRegistered(this);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).channelRegistered(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -190,12 +180,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeChannelUnregistered() {
|
private void invokeChannelUnregistered() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).channelUnregistered(this);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).channelUnregistered(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -220,12 +205,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeChannelActive() {
|
private void invokeChannelActive() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).channelActive(this);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).channelActive(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -250,12 +230,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeChannelInactive() {
|
private void invokeChannelInactive() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).channelInactive(this);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).channelInactive(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -293,12 +268,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeExceptionCaught(final Throwable cause) {
|
private void invokeExceptionCaught(final Throwable cause) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
handler.exceptionCaught(this, cause);
|
||||||
try {
|
|
||||||
handler.exceptionCaught(this, cause);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
if (logger.isWarnEnabled()) {
|
if (logger.isWarnEnabled()) {
|
||||||
logger.warn(
|
logger.warn(
|
||||||
@ -331,12 +301,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeUserEventTriggered(Object event) {
|
private void invokeUserEventTriggered(Object event) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).userEventTriggered(this, event);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).userEventTriggered(this, event);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -365,12 +330,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeChannelRead(Object msg) {
|
private void invokeChannelRead(Object msg) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).channelRead(this, msg);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).channelRead(this, msg);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -399,12 +359,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeChannelReadComplete() {
|
private void invokeChannelReadComplete() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).channelReadComplete(this);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).channelReadComplete(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -433,12 +388,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeChannelWritabilityChanged() {
|
private void invokeChannelWritabilityChanged() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelInboundHandler) handler).channelWritabilityChanged(this);
|
||||||
try {
|
|
||||||
((ChannelInboundHandler) handler).channelWritabilityChanged(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -499,12 +449,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
|
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).bind(this, localAddress, promise);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).bind(this, localAddress, promise);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyOutboundHandlerException(t, promise);
|
notifyOutboundHandlerException(t, promise);
|
||||||
}
|
}
|
||||||
@ -542,12 +487,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeConnect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
|
private void invokeConnect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).connect(this, remoteAddress, localAddress, promise);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).connect(this, remoteAddress, localAddress, promise);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyOutboundHandlerException(t, promise);
|
notifyOutboundHandlerException(t, promise);
|
||||||
}
|
}
|
||||||
@ -582,12 +522,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeDisconnect(ChannelPromise promise) {
|
private void invokeDisconnect(ChannelPromise promise) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).disconnect(this, promise);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).disconnect(this, promise);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyOutboundHandlerException(t, promise);
|
notifyOutboundHandlerException(t, promise);
|
||||||
}
|
}
|
||||||
@ -615,12 +550,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeClose(ChannelPromise promise) {
|
private void invokeClose(ChannelPromise promise) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).close(this, promise);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).close(this, promise);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyOutboundHandlerException(t, promise);
|
notifyOutboundHandlerException(t, promise);
|
||||||
}
|
}
|
||||||
@ -648,12 +578,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeDeregister(ChannelPromise promise) {
|
private void invokeDeregister(ChannelPromise promise) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).deregister(this, promise);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).deregister(this, promise);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyOutboundHandlerException(t, promise);
|
notifyOutboundHandlerException(t, promise);
|
||||||
}
|
}
|
||||||
@ -683,12 +608,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeRead() {
|
private void invokeRead() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).read(this);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).read(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
@ -724,12 +644,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeWrite(Object msg, ChannelPromise promise) {
|
private void invokeWrite(Object msg, ChannelPromise promise) {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).write(this, msg, promise);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).write(this, msg, promise);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyOutboundHandlerException(t, promise);
|
notifyOutboundHandlerException(t, promise);
|
||||||
}
|
}
|
||||||
@ -759,12 +674,7 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
|
|||||||
|
|
||||||
private void invokeFlush() {
|
private void invokeFlush() {
|
||||||
try {
|
try {
|
||||||
callDepth ++;
|
((ChannelOutboundHandler) handler).flush(this);
|
||||||
try {
|
|
||||||
((ChannelOutboundHandler) handler).flush(this);
|
|
||||||
} finally {
|
|
||||||
callDepth --;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
notifyHandlerException(t);
|
notifyHandlerException(t);
|
||||||
}
|
}
|
||||||
|
@ -506,16 +506,14 @@ final class DefaultChannelPipeline implements ChannelPipeline {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void callHandlerRemoved(final DefaultChannelHandlerContext ctx) {
|
private void callHandlerRemoved(final DefaultChannelHandlerContext ctx) {
|
||||||
if (ctx.channel().isRegistered()) {
|
if (ctx.channel().isRegistered() && !ctx.executor().inEventLoop()) {
|
||||||
if (!ctx.executor().inEventLoop() || ctx.isRunning()) {
|
ctx.executor().execute(new Runnable() {
|
||||||
ctx.executor().execute(new Runnable() {
|
|
||||||
@Override
|
@Override
|
||||||
public void run() {
|
public void run() {
|
||||||
callHandlerRemoved0(ctx);
|
callHandlerRemoved0(ctx);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
return;
|
return;
|
||||||
}
|
|
||||||
}
|
}
|
||||||
callHandlerRemoved0(ctx);
|
callHandlerRemoved0(ctx);
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user