title: Netty performs slow at writing
body: <p>Netty 4.1 (on OpenJDK 1.6.0_32 and CentOS 6.4) message sending is strangely slow. According to the profiler, it is the DefaultChannelHandlerContext.writeAndFlush that makes the biggest percentage (60%) of the running time. Decoding process is not emphasized in the profiler. Small messages are being processed and maybe the bootstrap options are not set correctly (TCP_NODELAY is true and nothing improved)? DefaultEventExecutorGroup is used both in server and client to avoid blocking Netty's main event loop and to run 'ServerData' and 'ClientData' classes with business logic and sending of the messages is done from there through context.writeAndFlush(...). Is there a more proper/faster way? Using straight ByteBuf.writeBytes(..) serialization in the encoder and ReplayingDecoder in the decoder made no difference in encoding speed. Sorry for the lengthy code, neither 'Netty In Action' book nor the documentation helped.</p>
  
  <p>JProfiler's call tree of the client side: <a href="http://i62.tinypic.com/dw4e43.jpg" rel="nofollow">http://i62.tinypic.com/dw4e43.jpg</a></p>
  
  <p>The server class is:</p>
  
  <pre><code>public class NettyServer 
  {
  EventLoopGroup incomingLoopGroup = null;
  EventLoopGroup workerLoopGroup = null;
  ServerBootstrap serverBootstrap = null;
  int port;
  DataServer dataServer = null;
  DefaultEventExecutorGroup dataEventExecutorGroup = null;    
  DefaultEventExecutorGroup dataEventExecutorGroup2 = null;   
  
  public ChannelFuture serverChannelFuture = null; 
  
  public NettyServer(int port)
  {
      this.port = port;
      DataServer = new DataServer(this);
  }
  
  public void run() throws Exception
  {
      incomingLoopGroup = new NioEventLoopGroup();
      workerLoopGroup = new NioEventLoopGroup();
  
      dataEventExecutorGroup = new DefaultEventExecutorGroup(5);
      dataEventExecutorGroup2 = new DefaultEventExecutorGroup(5);
  
      try
      {
          ChannelInitializer&lt;SocketChannel&gt; channelInitializer =
                  new ChannelInitializer&lt;SocketChannel&gt;()
                  {
                      @Override
                      protected void initChannel(SocketChannel ch)
                              throws Exception {                          
                          ch.pipeline().addLast(new MessageByteDecoder());
                          ch.pipeline().addLast(new MessageByteEncoder());
                          ch.pipeline().addLast(dataEventExecutorGroup, new DataServerInboundHandler(DataServer, NettyServer.this));
                          ch.pipeline().addLast(dataEventExecutorGroup2, new DataServerDataHandler(DataServer));
                      }
  
                  };
  
          // bootstrap the server
          serverBootstrap = new ServerBootstrap();
          serverBootstrap.group(incomingLoopGroup, workerLoopGroup)
              .channel(NioServerSocketChannel.class)
              .childHandler(channelInitializer)
              .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
              .option(ChannelOption.TCP_NODELAY, true)
              .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024)
              .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024)
              .childOption(ChannelOption.SO_KEEPALIVE, true);
  
          serverChannelFuture = serverBootstrap.bind(port).sync();
  
          serverChannelFuture.channel().closeFuture().sync();
  
      }
      finally
      {
          incomingLoopGroup.shutdownGracefully();
          workerLoopGroup.shutdownGracefully();
      }
  }
  
  }
  </code></pre>
  
  <p>The client class:</p>
  
  <pre><code>public class NettyClient 
  {
  Bootstrap clientBootstrap = null;
  EventLoopGroup workerLoopGroup = null;
  String serverHost = null;
  int serverPort = -1;
  
  ChannelFuture clientFutureChannel = null; 
  DataClient dataClient = null;
  DefaultEventExecutorGroup dataEventExecutorGroup = new DefaultEventExecutorGroup(5);    
  DefaultEventExecutorGroup dataEventExecutorGroup2 = new DefaultEventExecutorGroup(5);
  
  public NettyClient(String serverHost, int serverPort)
  {
      this.serverHost = serverHost;
      this.serverPort = serverPort;
  }
  
  public void run() throws Exception
  {
      workerLoopGroup = new NioEventLoopGroup();
  
      try
      {
          this.dataClient = new DataClient();
  
          ChannelInitializer&lt;SocketChannel&gt; channelInitializer =
                  new ChannelInitializer&lt;SocketChannel&gt;()
                  {
                      @Override
                      protected void initChannel(SocketChannel ch)
                              throws Exception {                          
                          ch.pipeline().addLast(new MessageByteDecoder());
                          ch.pipeline().addLast(new MessageByteEncoder());
                          ch.pipeline().addLast(dataEventExecutorGroup, new ClientInboundHandler(dataClient, NettyClient.this));                                          ch.pipeline().addLast(dataEventExecutorGroup2, new ClientDataHandler(dataClient));                          
                      }
  
                  };
  
          clientBootstrap = new Bootstrap();
          clientBootstrap.group(workerLoopGroup);
          clientBootstrap.channel(NioSocketChannel.class);
          clientBootstrap.option(ChannelOption.SO_KEEPALIVE, true);
          clientBootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
          clientBootstrap.option(ChannelOption.TCP_NODELAY, true);
          clientBootstrap.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
          clientBootstrap.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);
          clientBootstrap.handler(channelInitializer);
  
          clientFutureChannel = clientBootstrap.connect(serverHost, serverPort).sync();
          clientFutureChannel.channel().closeFuture().sync();
      }
      finally 
      {
          workerLoopGroup.shutdownGracefully();
      }
  }   
  
  }
  </code></pre>
  
  <p>The message class:</p>
  
  <pre><code>public class Message implements Serializable
  {
  public static final byte MSG_FIELD = 0;
  public static final byte MSG_HELLO = 1;
  public static final byte MSG_LOG = 2;
  public static final byte MSG_FIELD_RESPONSE = 3;
  public static final byte MSG_MAP_KEY_VALUE = 4;
  public static final byte MSG_STATS_FILE = 5;
  public static final byte MSG_SHUTDOWN = 6;
  
  public byte msgID;
  
  public byte msgType;     
  public String key;
  public String value;
  
  public byte method;
  public byte id;
  }
  </code></pre>
  
  <p>The decoder:</p>
  
  <pre><code>public class MessageByteDecoder extends ByteToMessageDecoder 
  {
  private Kryo kryoCodec = new Kryo();
  private int contentSize = 0;
  
  @Override
  protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List&lt;Object&gt; out) //throws Exception 
  {
      if (!buffer.isReadable() || buffer.readableBytes() &lt; 4) // we need at least integer
          return;
  
      // read header
      if (contentSize == 0) {
          contentSize = buffer.readInt();
      }
  
      if (buffer.readableBytes() &lt; contentSize) 
          return;
  
      // read content
      byte [] buf = new byte[contentSize];
      buffer.readBytes(buf);
  
      Input in = new Input(buf, 0, buf.length);
      out.add(kryoCodec.readObject(in, Message.class));
      contentSize = 0;
  }
  
  }
  </code></pre>
  
  <p>The encoder:</p>
  
  <pre><code>public class MessageByteEncoder extends MessageToByteEncoder&lt;Message&gt; 
  {
  Kryo kryoCodec = new Kryo();
  
  public MessageByteEncoder()
  {
      super(false);
  }
  
  @Override
  protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception 
  {       
      int offset = out.arrayOffset() + out.writerIndex();
      byte [] inArray = out.array();
      Output kryoOutput = new OutputWithOffset(inArray, inArray.length, offset + 4);
  
      // serialize message content
      kryoCodec.writeObject(kryoOutput, msg);
  
      // write length of the message content at the beginning of the array
      out.writeInt(kryoOutput.position());
      out.writerIndex(out.writerIndex() + kryoOutput.position());             
  }
  
  }
  </code></pre>
  
  <p>Client's business logic run in DefaultEventExecutorGroup:</p>
  
  <pre><code>public class DataClient
  {
  ChannelHandlerContext ctx;
  // ...
  public void processData()
  {
      // ...
      while ((line = br.readLine()) != null) 
      {
          // ...
          process = new CountDownLatch(columns.size());
          for(Column c : columns)
          {
              // sending column data to the server for processing
              ctx.channel().eventLoop().execute(new Runnable() {
                                   @Override
                                   public void run() {
                                          ctx.writeAndFlush(Message.createMessage(msgID, processID, c.key, c.value));               
            }});
  
          }
          // block until all the processed column fields of this row are returned from the server
          process.await();
          // write processed line to file ...
      }
      // ...
  }
  // ...
  }
  </code></pre>
  
  <p>Client's message handling:</p>
  
  <pre><code>public class ClientInboundHandler extends ChannelInboundHandlerAdapter 
  {
  DataClient dataClient = null;
  // ...
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg)
  {
      // dispatch the message to the listeners
      Message m = (Message) msg;
      switch(m.msgType)
      {
          case Message.MSG_FIELD_RESPONSE: // message with processed data is received from the server
              // decreases the 'process' CountDownLatch in the processData() method
              dataClient.setProcessingResult(m.msgID, m.value);
          break;
          // ...
      }       
      // forward the message to the pipeline
      ctx.fireChannelRead(msg);
  }
  // ...
  }
  }
  </code></pre>
  
  <p>Server's message handling:</p>
  
  <pre><code>public class ServerInboundHandler extends ChannelInboundHandlerAdapter 
  {
  private DataServer dataServer = null;
  // ...      
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object obj) throws Exception
  {
      Message msg = (Message) obj;
      switch(msg.msgType)
      {
          case Message.MSG_FIELD:
              dataServer.processField(msg, ctx);
              break;
          // ...
      }       
      ctx.fireChannelRead(msg);
  }
  //...
  }
  </code></pre>
  
  <p>Server's business logic run in DefaultEventExecutorGroup:</p>
  
  <pre><code>public class DataServer
  {   
  // ...
  public void processField(final Message msg, final ChannelHandlerContext context)
  {
      context.executor().submit(new Runnable()
      {
          @Override
          public void run() 
          {
              String processedValue = (String) processField(msg.key, msg.value);
              final Message responseToClient = Message.createResponseFieldMessage(msg.msgID, processedValue);
              // send processed data to the client
              context.channel().eventLoop().submit(new Runnable(){
                          @Override
                          public void run() {
                              context.writeAndFlush(responseToClient);
                          }
                      });
          }           
      });
  }
  
  // ...
  }
  </code></pre>