Repository: netty/netty
Issue: ChannelTrafficShapingHandler may corrupt inbound data stream
Labels: defect

As can be seen from the code of channelRead in AbstractTrafficShapingHandler, inbound data may be passed to the next handler in a scheduled Runnable.

However, it is reason to believe that 
inboud data that arrives later may be passed on before the Runnable is executed, and so the order in the inbound stream is not maintained.


Issue comments:
The following test demonstrates the problem. This is run with the netty 4.0.15.Final version of Netty.

``` java

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.traffic.ChannelTrafficShapingHandler;

import java.net.SocketAddress;
import java.util.zip.CRC32;

import org.junit.Assert;
import org.junit.Test;


public class TrafficTest {

    @Test
    public void testTrafficShaping() throws Exception {
        testTrafficShapingHandler(100, false);
        System.out.println("OK without ChannelTrafficShapingHandler");

        for (int i = 1; i <= 10; i++) {
            long sleep = i * 100;
            System.out.print("Test with " + sleep + "ms sleep..");
            testTrafficShapingHandler(sleep, true);
            System.out.println(" OK");
        }

    }

    public void testTrafficShapingHandler(long sleepWhenNotWritable, final boolean useTrafficShaping) throws Exception {

        final Crc32Sink sink = new Crc32Sink();

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.channel(NioServerSocketChannel.class).group(new NioEventLoopGroup()).localAddress("localhost", 0);

        serverBootstrap.childHandler(new ChannelInitializer<Channel>() {

            @Override
            protected void initChannel(Channel channel) throws Exception {
                if (useTrafficShaping) {
                    channel.pipeline().addLast(new ChannelTrafficShapingHandler(0, 128 * 1024));
                }
                channel.pipeline().addLast(sink);
            }
        });

        Bootstrap clientBootstrap = new Bootstrap();
        clientBootstrap.channel(NioSocketChannel.class).group(new NioEventLoopGroup());
        clientBootstrap.handler(new ChannelInboundHandlerAdapter());

        try {

            ChannelFuture cf = serverBootstrap.bind().sync();
            SocketAddress serverAddress = cf.channel().localAddress();

            System.out.println("Server address = " + serverAddress);

            Channel clientChannel = clientBootstrap.connect(serverAddress).sync().channel();

            byte[] data = new byte[512 * 1024];

            for (int i = 0; i < data.length; i++) {
                data[i] = (byte) (i % 31);
            }

            CRC32 crc32 = new CRC32();
            for (int i = 0; i < data.length; i++) {
                crc32.update(data[i]);
            }

            for (int i = 0; i < 256; i++) {
                ByteBuf slice = Unpooled.wrappedBuffer(data, 2048 * i, 2048);
                while (!clientChannel.isWritable()) {
                    Thread.sleep(sleepWhenNotWritable);
                }
                clientChannel.writeAndFlush(slice);
            }

            while (sink.getByteCount() < data.length) {
                Thread.sleep(100);
            }

            clientChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
            sink.getChannel().closeFuture().sync();

            Assert.assertEquals(data.length, sink.getByteCount());
            Assert.assertEquals(crc32.getValue(), sink.getCrc32());

        } finally {
            clientBootstrap.group().shutdownGracefully();
            serverBootstrap.group().shutdownGracefully();
        }

    }

    private static class Crc32Sink extends ChannelInboundHandlerAdapter {

        private final CRC32 crc32 = new CRC32();

        private volatile long byteCount = 0;

        private Channel channel;

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            channel = ctx.channel();
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf buf = (ByteBuf) msg;
            byteCount += buf.readableBytes();
            for (int i = buf.readerIndex(); i < buf.writerIndex(); i++) {
                crc32.update(buf.readByte());
            }

            buf.release();
        }

        long getCrc32() {
            return crc32.getValue();
        }

        long getByteCount() {
            return byteCount;
        }

        Channel getChannel() {
            return channel;
        }

    }

}

```

@fredericBregier Care to check?

I am having the same problem!

In AbstractTrafficShapingHandler.java, why are we scheduling fireChannelRead (inside channelRead method). We received a byte seq, we started processing it by passing it to pipeline and by scheduling we are just delaying the process of next handler. I don't think there is a need for delaying the next handler from processing the inbound data. When I remove that scheduling code it is working fine.

@trustin I had lost this one (sorry) but proposal seems ok for me.
However build seems in error: unstable for ByteBuf memory check since not released, but it seems totally unrelated to the patch, right ?

Fix merged...


Pull request review comments:

Patch that resolved the issue:
diff --git a/handler/src/main/java/io/netty/handler/traffic/AbstractTrafficShapingHandler.java b/handler/src/main/java/io/netty/handler/traffic/AbstractTrafficShapingHandler.java
index 7492886e2f6..e4f016048af 100644
--- a/handler/src/main/java/io/netty/handler/traffic/AbstractTrafficShapingHandler.java
+++ b/handler/src/main/java/io/netty/handler/traffic/AbstractTrafficShapingHandler.java
@@ -245,17 +245,6 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) throw
                     }
                     ctx.executor().schedule(reopenTask, wait,
                             TimeUnit.MILLISECONDS);
-                } else {
-                    // Create a Runnable to update the next handler in the chain. If one was create before it will
-                    // just be reused to limit object creation
-                    Runnable bufferUpdateTask = new Runnable() {
-                        @Override
-                        public void run() {
-                            ctx.fireChannelRead(msg);
-                        }
-                    };
-                    ctx.executor().schedule(bufferUpdateTask, wait, TimeUnit.MILLISECONDS);
-                    return;
                 }
             }
         }


Test patch: