All geek questions in one place

Netty ServerBootStrap or childOption option and effect of buffer size on speed

Netty seems to slow down the upload / download speed when the buffer size is increased beyond the default and / or the wrong location to override ( optionor childOption) in the ServerBootStrap object. This becomes even more noticeable when the connection has a long delay (~ 300 ms)

Setup :

The Netty client is located on MacOS and has default values ​​only . Using "Network Link Conditioner" with a DNS delay of 300 ms. Default values: SendBuffer / ReceiveBuffer / LowWaterMark / HighWaterMark - 128KB / 128KB / 32KB / 64KB.

The Netty server is located in Windows 8.1 and has the default SendBuffer / ReceiveBuffer / LowWaterMark / HighWaterMark - 64KB / 64KB / 32KB / 64KB.

Netty version 4.1.6 Final.

Measurement speed using wirehark using I / O graphs with settings: Y Axis → SUM (field Y), field Y → tcp.len

Devices are on the local network.

Result (speed values) :

Server-> Client Translation (setting SO_SNDBUF and low / high value watermark values ​​0.5 * SO_SNDBUF / SO_SNDBUF):

Setting location\SO_SNDBUF   | 64KB   | 128KB  | 1024KB
-----------------------------------------------------------------
option                       | 3.6MB/s| 3.6MB/s| 3.6MB/s
childOption                  | 0.2MB/s| 0.5MB/s| 3.6MB/s

Client-> Server Transfer (setting SO_RCVBUF):

Setting location\SO_RCVBUF   | 64KB   | 128KB  | 1024KB
-----------------------------------------------------------------
option                       | 0.2MB/s| 0.5MB/s| 3.6MB/s
childOption                  | 3.6MB/s| 0.4MB/s| 3.6MB/s

Server Code:

"args" values:

"sendToClient" / "sendToServer" (implicit) for sending direction. "option" / "childOption" for the override type. "1" / "2" / "3" for buffer values.

ObjectEchoServer.java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;

public final class ObjectEchoServer {

    static final int PORT = 8007;

    public static void main(String[] args) throws Exception {
        int rcvBuf, sndBuf, lowWaterMark, highWaterMark;
        rcvBuf = sndBuf = lowWaterMark = highWaterMark = 0;
        switch (args[2]){
            case "1":
                rcvBuf = 64;
                sndBuf = 64;
                lowWaterMark = 32;
                highWaterMark = 64;
                break;
            case "2":
                rcvBuf = 128;
                sndBuf = 128;
                lowWaterMark = 64;
                highWaterMark = 128;
                break;
            case "3":
                rcvBuf = 1024;
                sndBuf = 1024;
                lowWaterMark = 512;
                highWaterMark = 1024;
                break;

        }

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                            new ObjectEchoServerHandler(args[0]));
                }
             });
            if(args[1].equalsIgnoreCase("childOption")) {
                b.childOption(ChannelOption.SO_RCVBUF, rcvBuf * 1024);
                b.childOption(ChannelOption.SO_SNDBUF, sndBuf * 1024);
                b.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWaterMark * 1024, highWaterMark * 1024));
            } else if(args[1].equalsIgnoreCase("option")){
                b.option(ChannelOption.SO_RCVBUF, rcvBuf * 1024);
                b.option(ChannelOption.SO_SNDBUF, sndBuf * 1024);
                b.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWaterMark * 1024, highWaterMark * 1024));
            }
            // Bind and start to accept incoming connections.
            b.bind(PORT).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

ObjectEchoServerHandler.java

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class ObjectEchoServerHandler extends ChannelInboundHandlerAdapter {
    private Object msg;
    ChannelHandlerContext ctx;
    String sendToClient;

    public ObjectEchoServerHandler(String sendToClient){
        this.sendToClient = sendToClient;

    }

    @Override
    public void channelActive(ChannelHandlerContext ctx){
        this.ctx = ctx;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if(sendToClient.equalsIgnoreCase("sendToClient")) {//send a data stream to server
            this.msg = msg;
            ctx.writeAndFlush(msg).addListener(trafficGenerator);
        } //else receive a data stream from client
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    private final ChannelFutureListener trafficGenerator = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                ctx.writeAndFlush(msg).addListener(trafficGenerator);
            } else {
                future.cause().printStackTrace();
                future.channel().close();
            }
        }
    };
}

:

"args" :

"sendToClient" / "sendToServer" () .

ObjectEchoClient.java

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public final class ObjectEchoClient {

    static final String HOST = System.getProperty("host", "127.0.0.1");
    static final int PORT = 8007;
    static final int SIZE = 256;

    public static void main(String[] args) throws Exception {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                            new ObjectEchoClientHandler(args[0]));
                    System.out.println("senbuf:"+ ch.config().getSendBufferSize());
                    System.out.println("waterhigh:"+ ch.config().getWriteBufferWaterMark().high());
                    System.out.println("waterlow:"+ ch.config().getWriteBufferWaterMark().low());
                    System.out.println("recbuf:"+ ch.config().getReceiveBufferSize());

                }
             });
            // Start the connection attempt.
            b.connect(HOST, PORT).sync().channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

ObjectEchoClientHandler.java

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.ArrayList;
import java.util.List;

public class ObjectEchoClientHandler extends ChannelInboundHandlerAdapter {

    private final List<String> firstMessage;
    private ChannelHandlerContext ctx;
    private String sendToClient;

    ObjectEchoClientHandler(String sendToClient) {
        this.sendToClient = sendToClient;

        firstMessage = new ArrayList<>(ObjectEchoClient.SIZE);
        for (int i = 0; i < ObjectEchoClient.SIZE; i++) {
            firstMessage.add(Integer.toString(i));
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        this.ctx = ctx;
      if(sendToClient.equalsIgnoreCase("sendToClient")) {//get a data stream from server
          this.ctx.writeAndFlush(firstMessage);
      } else {//send a stream of to server
          this.ctx.writeAndFlush(firstMessage).addListener(trafficGenerator);
      }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ctx.flush();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    private final ChannelFutureListener trafficGenerator = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                ctx.writeAndFlush(firstMessage).addListener(trafficGenerator);
            } else {
                future.cause().printStackTrace();
                future.channel().close();
            }
        }
    };
}

:

/ ? , ( ), .

( ServerBootstrap.option() ServerBootstrap.childOption() netty 4.x) , ServerBootStrap.childOption , , ServerBootStrap.childOption.

+4

:

:

Source: https://habr.com/ru/post/1665731/

More articles:


All Articles