Clean up resources when shutdown

core/src/main/java/com/taobao/arthas/core/distribution/ResultConsumer.java

@@ -24,6 +24,10 @@ public interface ResultConsumer {
 
     long getLastAccessTime();
 
+    void close();
+
+    boolean isClosed();
+
     boolean isPolling();
 
     String getConsumerId();

core/src/main/java/com/taobao/arthas/core/distribution/ResultDistributor.java

@@ -14,4 +14,8 @@ public interface ResultDistributor {
      */
     void appendResult(ResultModel result);
 
+    /**
+     * Close result distribtor, release resources
+     */
+    void close();
 }

core/src/main/java/com/taobao/arthas/core/distribution/impl/PackingResultDistributorImpl.java

@@ -29,6 +29,10 @@ public class PackingResultDistributorImpl implements PackingResultDistributor {
         }
     }
 
+    @Override
+    public void close() {
+    }
+
     @Override
     public List<ResultModel> getResults() {
         ArrayList<ResultModel> results = new ArrayList<ResultModel>(resultQueue.size());

core/src/main/java/com/taobao/arthas/core/distribution/impl/ResultConsumerImpl.java

@@ -26,6 +26,7 @@ public class ResultConsumerImpl implements ResultConsumer {
     private int resultSizeLimit = 100;
     private long pollTimeLimit = 10 * 1000;
     private String consumerId;
+    private boolean closed;
 
     public ResultConsumerImpl() {
         lastAccessTime = System.currentTimeMillis();
@@ -51,7 +52,8 @@ public class ResultConsumerImpl implements ResultConsumer {
                 long waitingTime = 0;
                 List<ResultModel> sendingResults = new ArrayList<ResultModel>(resultSizeLimit);
 
-                while (sendingResults.size() < resultSizeLimit
+                while (!closed
+                        &&sendingResults.size() < resultSizeLimit
                         && sendingDelay < 200
                         && waitingTime < pollTimeLimit) {
                     ResultModel aResult = resultQueue.poll(100, TimeUnit.MILLISECONDS);
@@ -92,6 +94,16 @@ public class ResultConsumerImpl implements ResultConsumer {
         return lastAccessTime;
     }
 
+    @Override
+    public void close(){
+        this.closed = true;
+    }
+
+    @Override
+    public boolean isClosed() {
+        return closed;
+    }
+
     @Override
     public boolean isPolling() {
         return polling;

core/src/main/java/com/taobao/arthas/core/distribution/impl/SharingResultDistributorImpl.java

@@ -62,6 +62,11 @@ public class SharingResultDistributorImpl implements SharingResultDistributor {
         }
     }
 
+    @Override
+    public void close() {
+        this.running = false;
+    }
+
     @Override
     public void addConsumer(ResultConsumer consumer) {
         int consumerNo = consumerNumGenerator.incrementAndGet();
@@ -77,6 +82,7 @@ public class SharingResultDistributorImpl implements SharingResultDistributor {
     @Override
     public void removeConsumer(ResultConsumer consumer) {
         consumers.remove(consumer);
+        consumer.close();
     }
 
     @Override
@@ -153,6 +159,16 @@ public class SharingResultDistributorImpl implements SharingResultDistributor {
             return 0;
         }
 
+        @Override
+        public void close() {
+
+        }
+
+        @Override
+        public boolean isClosed() {
+            return false;
+        }
+
         @Override
         public boolean isPolling() {
             return false;

core/src/main/java/com/taobao/arthas/core/distribution/impl/TermResultDistributorImpl.java

@@ -29,4 +29,8 @@ public class TermResultDistributorImpl implements ResultDistributor {
         }
     }
 
+    @Override
+    public void close() {
+    }
+
 }

core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java

@@ -73,6 +73,7 @@ public class ArthasBootstrap {
     private File arthasOutputDir;
 
     private static LoggerContext loggerContext;
+    private EventExecutorGroup workerGroup;
 
     private ArthasBootstrap(Instrumentation instrumentation, String args) throws Throwable {
         this.instrumentation = instrumentation;
@@ -269,7 +270,7 @@ public class ArthasBootstrap {
             resolvers.add(builtinCommands);
 
             //worker group
-            EventExecutorGroup workerGroup = new NioEventLoopGroup(8);
+            workerGroup = new NioEventLoopGroup(24);
 
             // TODO: discover user provided command resolver
             if (configure.getTelnetPort() > 0) {
@@ -313,10 +314,18 @@ public class ArthasBootstrap {
             if (sessionManager != null){
                 sessionManager.close();
             }
+            shutdownWorkGroup();
             throw e;
         }
     }
 
+    private void shutdownWorkGroup() {
+        if (workerGroup != null) {
+            workerGroup.shutdownGracefully(200, 200, TimeUnit.MILLISECONDS);
+            workerGroup = null;
+        }
+    }
+
     /**
      * 判断服务端是否已经启动
      *
@@ -336,6 +345,7 @@ public class ArthasBootstrap {
         }
         executorService.shutdownNow();
         UserStatUtil.destroy();
+        shutdownWorkGroup();
         // clear the reference in Spy class.
         cleanUpSpyReference();
         try {
@@ -347,6 +357,8 @@ public class ArthasBootstrap {
         if (loggerContext != null) {
             loggerContext.stop();
         }
+        shellServer = null;
+        sessionManager = null;
     }
 
     /**

core/src/main/java/com/taobao/arthas/core/shell/session/impl/SessionManagerImpl.java

@@ -93,7 +93,9 @@ public class SessionManagerImpl implements SessionManager {
 
         ArrayList<Session> sessions = new ArrayList<Session>(this.sessions.values());
         for (Session session : sessions) {
-            session.getResultDistributor().appendResult(new MessageModel("arthas server is going to shutdown."));
+            SharingResultDistributor resultDistributor = session.getResultDistributor();
+            resultDistributor.appendResult(new MessageModel("arthas server is going to shutdown."));
+            resultDistributor.close();
             logger.info("Removing session before shutdown: {}, last access time: {}", session.getSessionId(), session.getLastAccessTime());
             this.removeSession(session.getSessionId());
         }