From 8fa9ca592ce526ffe191ec4e9c6889997e5bfc93 Mon Sep 17 00:00:00 2001 From: huangli Date: Thu, 10 Feb 2022 11:15:04 +0800 Subject: [PATCH] [ISSUE 3825] Use default SO_SNDBUF/SO_RCVBUF/WRITE_BUFFER_WATER_MARK value. (#3826) This commit may greatly increase thoughtput when network lantency is big or message size is big, especially for consumer. In one broker (M/S sync master async flush), one producer send test. 250 byte body: 0.93x old version 4000 byte body: 3.85x of old version 16000 byte body: 6.07x of old version 100000 byte body: 7.93x of old version The old version is 4.9.3-SNAPSHOT in this test. In this test the network latency from producer to broker is 1.61ms. --- .../apache/rocketmq/broker/BrokerStartup.java | 9 --------- .../rocketmq/remoting/common/RemotingUtil.java | 9 +++++++-- .../remoting/netty/NettyRemotingClient.java | 18 ++++++++++++++---- .../remoting/netty/NettyRemotingServer.java | 18 ++++++++++++++---- .../remoting/netty/NettySystemConfig.java | 8 ++++---- .../apache/rocketmq/store/ha/HAConnection.java | 9 +++++++-- 6 files changed, 46 insertions(+), 25 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerStartup.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerStartup.java index bb2eb9135a..19e618ba2d 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerStartup.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerStartup.java @@ -32,7 +32,6 @@ import org.apache.rocketmq.remoting.common.RemotingUtil; import org.apache.rocketmq.remoting.common.TlsMode; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.netty.NettyServerConfig; -import org.apache.rocketmq.remoting.netty.NettySystemConfig; import org.apache.rocketmq.remoting.netty.TlsSystemConfig; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.srvutil.ServerUtil; @@ -90,14 +89,6 @@ public class BrokerStartup { public static BrokerController createBrokerController(String[] args) { System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION)); - if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) { - NettySystemConfig.socketSndbufSize = 131072; - } - - if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) { - NettySystemConfig.socketRcvbufSize = 131072; - } - try { //PackageConflictDetect.detectFastjson(); Options options = ServerUtil.buildCommandlineOptions(new Options()); diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java index 3914314f0e..d5ce20b0f1 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java @@ -34,6 +34,7 @@ import java.util.ArrayList; import java.util.Enumeration; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; +import org.apache.rocketmq.remoting.netty.NettySystemConfig; public class RemotingUtil { public static final String OS_NAME = System.getProperty("os.name"); @@ -193,8 +194,12 @@ public class RemotingUtil { sc.configureBlocking(true); sc.socket().setSoLinger(false, -1); sc.socket().setTcpNoDelay(true); - sc.socket().setReceiveBufferSize(1024 * 64); - sc.socket().setSendBufferSize(1024 * 64); + if (NettySystemConfig.socketSndbufSize > 0) { + sc.socket().setReceiveBufferSize(NettySystemConfig.socketSndbufSize); + } + if (NettySystemConfig.socketRcvbufSize > 0) { + sc.socket().setSendBufferSize(NettySystemConfig.socketRcvbufSize); + } sc.socket().connect(remote, timeoutMillis); sc.configureBlocking(false); return sc; diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java index cbd17fadb2..5ced3b7eb4 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java @@ -166,10 +166,6 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.SO_KEEPALIVE, false) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis()) - .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize()) - .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize()) - .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(nettyClientConfig.getWriteBufferLowWaterMark(), - nettyClientConfig.getWriteBufferHighWaterMark())) .handler(new ChannelInitializer() { @Override public void initChannel(SocketChannel ch) throws Exception { @@ -191,6 +187,20 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti new NettyClientHandler()); } }); + if (nettyClientConfig.getClientSocketSndBufSize() > 0) { + log.info("client set SO_SNDBUF to {}", nettyClientConfig.getClientSocketSndBufSize()); + handler.option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize()); + } + if (nettyClientConfig.getClientSocketRcvBufSize() > 0) { + log.info("client set SO_RCVBUF to {}", nettyClientConfig.getClientSocketRcvBufSize()); + handler.option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize()); + } + if (nettyClientConfig.getWriteBufferLowWaterMark() > 0 && nettyClientConfig.getWriteBufferHighWaterMark() > 0) { + log.info("client set netty WRITE_BUFFER_WATER_MARK to {},{}", + nettyClientConfig.getWriteBufferLowWaterMark(), nettyClientConfig.getWriteBufferHighWaterMark()); + handler.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark( + nettyClientConfig.getWriteBufferLowWaterMark(), nettyClientConfig.getWriteBufferHighWaterMark())); + } this.timer.scheduleAtFixedRate(new TimerTask() { @Override diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java index 2b7413f7a9..22440af943 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java @@ -203,10 +203,6 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti .option(ChannelOption.SO_REUSEADDR, true) .option(ChannelOption.SO_KEEPALIVE, false) .childOption(ChannelOption.TCP_NODELAY, true) - .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize()) - .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize()) - .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark( - nettyServerConfig.getWriteBufferLowWaterMark(), nettyServerConfig.getWriteBufferHighWaterMark())) .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort())) .childHandler(new ChannelInitializer() { @Override @@ -222,6 +218,20 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti ); } }); + if (nettyServerConfig.getServerSocketSndBufSize() > 0) { + log.info("server set SO_SNDBUF to {}", nettyServerConfig.getServerSocketSndBufSize()); + childHandler.childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize()); + } + if (nettyServerConfig.getServerSocketRcvBufSize() > 0) { + log.info("server set SO_RCVBUF to {}", nettyServerConfig.getServerSocketRcvBufSize()); + childHandler.childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize()); + } + if (nettyServerConfig.getWriteBufferLowWaterMark() > 0 && nettyServerConfig.getWriteBufferHighWaterMark() > 0) { + log.info("server set netty WRITE_BUFFER_WATER_MARK to {},{}", + nettyServerConfig.getWriteBufferLowWaterMark(), nettyServerConfig.getWriteBufferHighWaterMark()); + childHandler.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark( + nettyServerConfig.getWriteBufferLowWaterMark(), nettyServerConfig.getWriteBufferHighWaterMark())); + } if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) { childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java index 4290baea7a..edad8444ed 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java @@ -50,9 +50,9 @@ public class NettySystemConfig { public static final int CLIENT_ONEWAY_SEMAPHORE_VALUE = Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_ONEWAY_SEMAPHORE_VALUE, "65535")); public static int socketSndbufSize = - Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE, "65535")); + Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE, "0")); public static int socketRcvbufSize = - Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE, "65535")); + Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE, "0")); public static int socketBacklog = Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_SOCKET_BACKLOG, "1024")); public static int clientWorkerSize = @@ -64,8 +64,8 @@ public class NettySystemConfig { public static boolean clientCloseSocketIfTimeout = Boolean.parseBoolean(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_CLOSE_SOCKET_IF_TIMEOUT, "true")); public static int writeBufferHighWaterMark = - Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_WRITE_BUFFER_HIGH_WATER_MARK_VALUE, "4194304"));//4M + Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_WRITE_BUFFER_HIGH_WATER_MARK_VALUE, "0")); public static int writeBufferLowWaterMark = - Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_WRITE_BUFFER_LOW_WATER_MARK, "1048576")); //1MB + Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_WRITE_BUFFER_LOW_WATER_MARK, "0")); } diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java index c08c515c0c..4c26971c07 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java +++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java @@ -26,6 +26,7 @@ import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.remoting.common.RemotingUtil; +import org.apache.rocketmq.remoting.netty.NettySystemConfig; import org.apache.rocketmq.store.SelectMappedBufferResult; public class HAConnection { @@ -46,8 +47,12 @@ public class HAConnection { this.socketChannel.configureBlocking(false); this.socketChannel.socket().setSoLinger(false, -1); this.socketChannel.socket().setTcpNoDelay(true); - this.socketChannel.socket().setReceiveBufferSize(1024 * 64); - this.socketChannel.socket().setSendBufferSize(1024 * 64); + if (NettySystemConfig.socketSndbufSize > 0) { + this.socketChannel.socket().setReceiveBufferSize(NettySystemConfig.socketSndbufSize); + } + if (NettySystemConfig.socketRcvbufSize > 0) { + this.socketChannel.socket().setSendBufferSize(NettySystemConfig.socketRcvbufSize); + } this.writeSocketService = new WriteSocketService(this.socketChannel); this.readSocketService = new ReadSocketService(this.socketChannel); this.haService.getConnectionCount().incrementAndGet();