From 898da3da022eb6a0165d64cf72cd678cf0cd028f Mon Sep 17 00:00:00 2001 From: gongdewei Date: Thu, 21 May 2020 22:11:28 +0800 Subject: [PATCH] Use custom worker group to reduce the occurrence of http api block --- .../arthas/core/server/ArthasBootstrap.java | 56 +++++++++---------- .../core/shell/term/impl/HttpTermServer.java | 8 ++- .../impl/http/NettyWebsocketTtyBootstrap.java | 7 ++- .../term/impl/http/TtyServerInitializer.java | 12 ++-- .../impl/httptelnet/HttpTelnetTermServer.java | 7 ++- .../httptelnet/NettyHttpTelnetBootstrap.java | 7 ++- .../NettyHttpTelnetTtyBootstrap.java | 5 +- .../httptelnet/ProtocolDetectHandler.java | 9 ++- 8 files changed, 64 insertions(+), 47 deletions(-) diff --git a/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java b/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java index 6aa9b6fe3..27823c526 100644 --- a/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java +++ b/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java @@ -48,27 +48,10 @@ import com.taobao.arthas.core.util.ArthasBanner; import com.taobao.arthas.core.util.FileUtils; import com.taobao.arthas.core.util.LogUtil; import com.taobao.arthas.core.util.UserStatUtil; - import io.netty.channel.ChannelFuture; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.util.concurrent.EventExecutorGroup; -import java.arthas.SpyAPI; -import java.io.File; -import java.io.IOException; -import java.lang.instrument.Instrumentation; -import java.lang.reflect.Method; -import java.net.URI; -import java.security.CodeSource; -import java.util.*; -import java.util.Map.Entry; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; - - /** * @author vlinux on 15/5/2. @@ -78,9 +61,9 @@ public class ArthasBootstrap { public static final String ARTHAS_HOME_PROPERTY = "arthas.home"; private static String ARTHAS_SHOME = null; - public static final String CONFIG_NAME_PROPERTY = "arthas.config.name"; + public static final String CONFIG_NAME_PROPERTY = "arthas.config.name"; public static final String CONFIG_LOCATION_PROPERTY = "arthas.config.location"; - public static final String CONFIG_OVERRIDE_ALL= "arthas.config.overrideAll"; + public static final String CONFIG_OVERRIDE_ALL = "arthas.config.overrideAll"; private static ArthasBootstrap arthasBootstrap; @@ -98,6 +81,7 @@ public class ArthasBootstrap { private File arthasOutputDir; private static LoggerContext loggerContext; + private EventExecutorGroup workerGroup; private Timer timer = new Timer("arthas-timer", true); @@ -258,14 +242,14 @@ public class ArthasBootstrap { tunnelClient.setTunnelServerUrl(configure.getTunnelServer()); // ws://127.0.0.1:8563/ws String host = "127.0.0.1"; - if(configure.getIp() != null) { + if (configure.getIp() != null) { host = configure.getIp(); } URI uri = new URI("ws", null, host, configure.getHttpPort(), "/ws", null, null); tunnelClient.setLocalServerUrl(uri.toString()); ChannelFuture channelFuture = tunnelClient.start(); channelFuture.await(10, TimeUnit.SECONDS); - if(channelFuture.isSuccess()) { + if (channelFuture.isSuccess()) { agentId = tunnelClient.getId(); } } @@ -275,29 +259,34 @@ public class ArthasBootstrap { try { ShellServerOptions options = new ShellServerOptions() - .setInstrumentation(instrumentation) - .setPid(PidUtils.currentLongPid()) - .setSessionTimeout(configure.getSessionTimeout() * 1000); + .setInstrumentation(instrumentation) + .setPid(PidUtils.currentLongPid()) + .setSessionTimeout(configure.getSessionTimeout() * 1000); if (agentId != null) { Map welcomeInfos = new HashMap(); welcomeInfos.put("id", agentId); options.setWelcomeMessage(ArthasBanner.welcome(welcomeInfos)); } + shellServer = new ShellServerImpl(options, this); BuiltinCommandPack builtinCommands = new BuiltinCommandPack(); List resolvers = new ArrayList(); resolvers.add(builtinCommands); + + //worker group + workerGroup = new NioEventLoopGroup(24); + // TODO: discover user provided command resolver if (configure.getTelnetPort() > 0) { shellServer.registerTermServer(new HttpTelnetTermServer(configure.getIp(), configure.getTelnetPort(), - options.getConnectionTimeout())); + options.getConnectionTimeout(), workerGroup)); } else { logger().info("telnet port is {}, skip bind telnet server.", configure.getTelnetPort()); } if (configure.getHttpPort() > 0) { shellServer.registerTermServer(new HttpTermServer(configure.getIp(), configure.getHttpPort(), - options.getConnectionTimeout())); + options.getConnectionTimeout(), workerGroup)); } else { logger().info("http port is {}, skip bind http server.", configure.getHttpPort()); } @@ -321,7 +310,7 @@ public class ArthasBootstrap { UserStatUtil.setStatUrl(configure.getStatUrl()); UserStatUtil.arthasStart(); - logger().info("as-server started in {} ms", System.currentTimeMillis() - start ); + logger().info("as-server started in {} ms", System.currentTimeMillis() - start); } catch (Throwable e) { logger().error("Error during bind to port " + configure.getTelnetPort(), e); if (shellServer != null) { @@ -330,11 +319,18 @@ public class ArthasBootstrap { if (sessionManager != null){ sessionManager.close(); } - //shutdownWorkGroup(); + shutdownWorkGroup(); throw e; } } + private void shutdownWorkGroup() { + if (workerGroup != null) { + workerGroup.shutdownGracefully(200, 200, TimeUnit.MILLISECONDS); + workerGroup = null; + } + } + /** * 判断服务端是否已经启动 * @@ -350,12 +346,13 @@ public class ArthasBootstrap { try { tunnelClient.stop(); } catch (Throwable e) { - logger().error("arthas", "stop tunnel client error", e); + logger().error("stop tunnel client error", e); } } executorService.shutdownNow(); transformerManager.destroy(); UserStatUtil.destroy(); + shutdownWorkGroup(); // clear the reference in Spy class. cleanUpSpyReference(); try { @@ -384,6 +381,7 @@ public class ArthasBootstrap { } return arthasBootstrap; } + /** * @return ArthasServer单例 */ diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/HttpTermServer.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/HttpTermServer.java index 2ad5f6288..c8dafe942 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/HttpTermServer.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/HttpTermServer.java @@ -7,7 +7,7 @@ import com.taobao.arthas.core.shell.handlers.Handler; import com.taobao.arthas.core.shell.term.Term; import com.taobao.arthas.core.shell.term.TermServer; import com.taobao.arthas.core.shell.term.impl.http.NettyWebsocketTtyBootstrap; -import com.taobao.arthas.core.shell.term.impl.httptelnet.HttpTelnetTermServer; +import io.netty.util.concurrent.EventExecutorGroup; import io.termd.core.function.Consumer; import io.termd.core.tty.TtyConnection; @@ -25,11 +25,13 @@ public class HttpTermServer extends TermServer { private String hostIp; private int port; private long connectionTimeout; + private EventExecutorGroup workerGroup; - public HttpTermServer(String hostIp, int port, long connectionTimeout) { + public HttpTermServer(String hostIp, int port, long connectionTimeout, EventExecutorGroup workerGroup) { this.hostIp = hostIp; this.port = port; this.connectionTimeout = connectionTimeout; + this.workerGroup = workerGroup; } @Override @@ -41,7 +43,7 @@ public class HttpTermServer extends TermServer { @Override public TermServer listen(Handler> listenHandler) { // TODO: charset and inputrc from options - bootstrap = new NettyWebsocketTtyBootstrap().setHost(hostIp).setPort(port); + bootstrap = new NettyWebsocketTtyBootstrap(workerGroup).setHost(hostIp).setPort(port); try { bootstrap.start(new Consumer() { @Override diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java index fefffce62..e6f5fd791 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java @@ -10,6 +10,7 @@ import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; +import io.netty.util.concurrent.EventExecutorGroup; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener; import io.netty.util.concurrent.ImmediateEventExecutor; @@ -30,8 +31,10 @@ public class NettyWebsocketTtyBootstrap { private int port; private EventLoopGroup group; private Channel channel; + private EventExecutorGroup workerGroup; - public NettyWebsocketTtyBootstrap() { + public NettyWebsocketTtyBootstrap(EventExecutorGroup workerGroup) { + this.workerGroup = workerGroup; this.host = "localhost"; this.port = 8080; } @@ -59,7 +62,7 @@ public class NettyWebsocketTtyBootstrap { ServerBootstrap b = new ServerBootstrap(); b.group(group).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)) - .childHandler(new TtyServerInitializer(channelGroup, handler)); + .childHandler(new TtyServerInitializer(channelGroup, handler, workerGroup)); final ChannelFuture f = b.bind(host, port); f.addListener(new GenericFutureListener>() { diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/TtyServerInitializer.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/TtyServerInitializer.java index 1dacb1f7a..57ba3c9a6 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/TtyServerInitializer.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/TtyServerInitializer.java @@ -1,7 +1,5 @@ package com.taobao.arthas.core.shell.term.impl.http; -import java.io.File; - import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.group.ChannelGroup; @@ -10,9 +8,12 @@ import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.stream.ChunkedWriteHandler; +import io.netty.util.concurrent.EventExecutorGroup; import io.termd.core.function.Consumer; import io.termd.core.tty.TtyConnection; +import java.io.File; + /** * @author Julien Viet @@ -21,19 +22,22 @@ public class TtyServerInitializer extends ChannelInitializer { private final ChannelGroup group; private final Consumer handler; + private EventExecutorGroup workerGroup; - public TtyServerInitializer(ChannelGroup group, Consumer handler) { + public TtyServerInitializer(ChannelGroup group, Consumer handler, EventExecutorGroup workerGroup) { this.group = group; this.handler = handler; + this.workerGroup = workerGroup; } @Override protected void initChannel(SocketChannel ch) throws Exception { + ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new HttpServerCodec()); pipeline.addLast(new ChunkedWriteHandler()); pipeline.addLast(new HttpObjectAggregator(64 * 1024)); - pipeline.addLast(new HttpRequestHandler("/ws", new File("arthas-output"))); + pipeline.addLast(workerGroup, "HttpRequestHandler", new HttpRequestHandler("/ws", new File("arthas-output"))); pipeline.addLast(new WebSocketServerProtocolHandler("/ws")); pipeline.addLast(new TtyWebSocketFrameHandler(group, handler)); } diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/HttpTelnetTermServer.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/HttpTelnetTermServer.java index 3bc6c5b66..f7de93789 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/HttpTelnetTermServer.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/HttpTelnetTermServer.java @@ -11,6 +11,7 @@ import com.taobao.arthas.core.shell.term.TermServer; import com.taobao.arthas.core.shell.term.impl.Helper; import com.taobao.arthas.core.shell.term.impl.TermImpl; +import io.netty.util.concurrent.EventExecutorGroup; import io.termd.core.function.Consumer; import io.termd.core.tty.TtyConnection; @@ -29,11 +30,13 @@ public class HttpTelnetTermServer extends TermServer { private String hostIp; private int port; private long connectionTimeout; + private EventExecutorGroup workerGroup; - public HttpTelnetTermServer(String hostIp, int port, long connectionTimeout) { + public HttpTelnetTermServer(String hostIp, int port, long connectionTimeout, EventExecutorGroup workerGroup) { this.hostIp = hostIp; this.port = port; this.connectionTimeout = connectionTimeout; + this.workerGroup = workerGroup; } @Override @@ -45,7 +48,7 @@ public class HttpTelnetTermServer extends TermServer { @Override public TermServer listen(Handler> listenHandler) { // TODO: charset and inputrc from options - bootstrap = new NettyHttpTelnetTtyBootstrap().setHost(hostIp).setPort(port); + bootstrap = new NettyHttpTelnetTtyBootstrap(workerGroup).setHost(hostIp).setPort(port); try { bootstrap.start(new Consumer() { @Override diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetBootstrap.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetBootstrap.java index b289582f5..a538e3266 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetBootstrap.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetBootstrap.java @@ -11,6 +11,7 @@ import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; +import io.netty.util.concurrent.EventExecutorGroup; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener; import io.netty.util.concurrent.ImmediateEventExecutor; @@ -28,8 +29,10 @@ public class NettyHttpTelnetBootstrap extends TelnetBootstrap { private EventLoopGroup group; private ChannelGroup channelGroup; + private EventExecutorGroup workerGroup; - public NettyHttpTelnetBootstrap() { + public NettyHttpTelnetBootstrap(EventExecutorGroup workerGroup) { + this.workerGroup = workerGroup; this.group = new NioEventLoopGroup(); this.channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE); } @@ -56,7 +59,7 @@ public class NettyHttpTelnetBootstrap extends TelnetBootstrap { .childHandler(new ChannelInitializer() { @Override public void initChannel(SocketChannel ch) throws Exception { - ch.pipeline().addLast(new ProtocolDetectHandler(channelGroup, handlerFactory, factory)); + ch.pipeline().addLast(new ProtocolDetectHandler(channelGroup, handlerFactory, factory, workerGroup)); } }); diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetTtyBootstrap.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetTtyBootstrap.java index 100c003f4..05a7ec08a 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetTtyBootstrap.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetTtyBootstrap.java @@ -2,6 +2,7 @@ package com.taobao.arthas.core.shell.term.impl.httptelnet; import java.nio.charset.Charset; +import io.netty.util.concurrent.EventExecutorGroup; import io.termd.core.function.Consumer; import io.termd.core.function.Supplier; import io.termd.core.telnet.TelnetHandler; @@ -21,8 +22,8 @@ public class NettyHttpTelnetTtyBootstrap { private boolean inBinary; private Charset charset = Charset.forName("UTF-8"); - public NettyHttpTelnetTtyBootstrap() { - this.httpTelnetTtyBootstrap = new NettyHttpTelnetBootstrap(); + public NettyHttpTelnetTtyBootstrap(EventExecutorGroup workerGroup) { + this.httpTelnetTtyBootstrap = new NettyHttpTelnetBootstrap(workerGroup); } public String getHost() { diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/ProtocolDetectHandler.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/ProtocolDetectHandler.java index 4ca8ea6f5..8113ef9fc 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/ProtocolDetectHandler.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/ProtocolDetectHandler.java @@ -4,8 +4,8 @@ import java.io.File; import java.util.concurrent.TimeUnit; import com.taobao.arthas.core.shell.term.impl.http.HttpRequestHandler; -import com.taobao.arthas.core.shell.term.impl.http.TtyWebSocketFrameHandler; +import com.taobao.arthas.core.shell.term.impl.http.TtyWebSocketFrameHandler; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; @@ -15,6 +15,7 @@ import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.stream.ChunkedWriteHandler; +import io.netty.util.concurrent.EventExecutorGroup; import io.netty.util.concurrent.ScheduledFuture; import io.termd.core.function.Consumer; import io.termd.core.function.Supplier; @@ -31,12 +32,14 @@ public class ProtocolDetectHandler extends ChannelInboundHandlerAdapter { private ChannelGroup channelGroup; private Supplier handlerFactory; private Consumer ttyConnectionFactory; + private EventExecutorGroup workerGroup; public ProtocolDetectHandler(ChannelGroup channelGroup, final Supplier handlerFactory, - Consumer ttyConnectionFactory) { + Consumer ttyConnectionFactory, EventExecutorGroup workerGroup) { this.channelGroup = channelGroup; this.handlerFactory = handlerFactory; this.ttyConnectionFactory = ttyConnectionFactory; + this.workerGroup = workerGroup; } private ScheduledFuture detectTelnetFuture; @@ -82,7 +85,7 @@ public class ProtocolDetectHandler extends ChannelInboundHandlerAdapter { pipeline.addLast(new HttpServerCodec()); pipeline.addLast(new ChunkedWriteHandler()); pipeline.addLast(new HttpObjectAggregator(64 * 1024)); - pipeline.addLast(new HttpRequestHandler("/ws", new File("arthas-output"))); + pipeline.addLast(workerGroup, "HttpRequestHandler", new HttpRequestHandler("/ws", new File("arthas-output"))); pipeline.addLast(new WebSocketServerProtocolHandler("/ws")); pipeline.addLast(new TtyWebSocketFrameHandler(channelGroup, ttyConnectionFactory)); ctx.fireChannelActive();