Update InlineServer shutdown code and spin a process if thread leak is detected

Author: myronkscott

common/src/main/java/com/tc/lang/TCThreadGroup.java

@@ -46,7 +46,7 @@ public TCThreadGroup(ThrowableHandler throwableHandler, String name) {
   }
 
   public TCThreadGroup(ThrowableHandler throwableHandler, String name, boolean stoppable) {
-    this(throwableHandler, name, stoppable, !stoppable);
+    this(throwableHandler, name, stoppable, false);
   }
 
   public TCThreadGroup(ThrowableHandler throwableHandler, String name, boolean stoppable, boolean ignorePool) {
@@ -76,6 +76,7 @@ public boolean isStoppable() {
   public void printLiveThreads(Consumer<String> reporter) {
     for (Thread t : threads()) {
       if (t != null && t != Thread.currentThread()) {
+        reporter.accept(t.getThreadGroup().getName() + " - " + t.getName());
         reporter.accept(ThreadDumpUtil.getThreadDump(t));
       }
     }
@@ -121,7 +122,7 @@ private boolean lookForThreadExit(Thread t, Consumer<InterruptedException> inter
     }
   }
 
-  private synchronized List<Thread> threads() {
+  private List<Thread> threads() {
     int ac = activeCount();
     Thread[] list = new Thread[ac];
     enumerate(list, true);

common/src/main/java/com/tc/net/protocol/tcm/CommunicationsManagerImpl.java

@@ -54,7 +54,6 @@
 import com.tc.net.protocol.transport.WireProtocolMessageSink;
 import com.tc.net.core.ProductID;
 import com.tc.util.Assert;
-import com.tc.util.TCTimeoutException;
 import com.tc.util.concurrent.SetOnceFlag;
 
 import java.io.IOException;

galvan-support/src/main/java/org/terracotta/testing/rules/BasicExternalCluster.java

@@ -247,7 +247,7 @@ private void internalStart(CompletableFuture<Void> checker) throws Throwable {
       ServerInstance serverProcess = !inline ?
         new ServerProcess(serverName, server, serverWorkingDir, serverHeapSize, debugPort, systemProperties, parentOutput, builder.build())
               :
-        new InlineServer(serverName, server, serverWorkingDir, systemProperties, parentOutput, builder.build());
+        new InlineServer(serverName, server, serverWorkingDir, serverHeapSize, debugPort, systemProperties, parentOutput, builder.build());
 
       stripeInstaller.installNewServer(serverProcess);
     }

galvan-support/src/test/java/org/terracotta/testing/rules/SimpleActivePassiveWithClassRuleIT.java

@@ -20,7 +20,6 @@
 
 import org.junit.ClassRule;
 import org.junit.Test;
-import org.terracotta.passthrough.IClusterControl;
 
 
 /**

galvan/src/main/java/org/terracotta/testing/master/IGalvanServer.java

@@ -39,4 +39,6 @@ public interface IGalvanServer {
   void waitForTermination();
 
   void waitForState(ServerMode mode) throws InterruptedException;
+  
+  IGalvanServer newInstance();
 }

galvan/src/main/java/org/terracotta/testing/master/InlineServer.java

@@ -28,11 +28,14 @@
 import java.nio.file.Path;
 import static java.nio.file.StandardOpenOption.APPEND;
 import static java.nio.file.StandardOpenOption.CREATE;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.terracotta.testing.logging.VerboseOutputStream;
@@ -47,16 +50,26 @@ public class InlineServer extends ServerInstance {
   private final OutputStream parentOutput;
   private final String[] cmd;
 
-  private ServerThread server;
 
-  public InlineServer(String serverName, Path serverInstall, Path serverWorkingDir, Properties serverProperties, OutputStream out, String[] cmd) {
+  private final int heap;
+  private final int debug;
+  private boolean leakDetected = false;
+  private ServerThread serverThread;
+
+  public InlineServer(String serverName, Path serverInstall, Path serverWorkingDir, int heap, int debug, Properties serverProperties, OutputStream out, String[] cmd) {
     super(serverName);
     // We need to specify a positive integer as the heap size.
     this.serverInstall = serverInstall;
     this.serverWorkingDir = serverWorkingDir;
     this.serverProperties = serverProperties;
     this.parentOutput = out;
     this.cmd = cmd;
+    this.debug = debug;
+    this.heap = heap;
+  }
+
+  public InlineServer(String serverName, Path serverInstall, Path serverWorkingDir, Properties serverProperties, OutputStream out, String[] cmd) {
+    this(serverName, serverInstall, serverWorkingDir, 512, 0, serverProperties, out, cmd);
   }
 
   /**
@@ -76,10 +89,10 @@ public void start() throws IOException {
     try {
       // First thing we need to do is make sure that we aren't already running.
       Assert.assertFalse(this.isServerRunning());
-      Assert.assertTrue(server == null || server.shutdown());
+      Assert.assertTrue(serverThread == null || serverThread.shutdown());
       setCurrentState(ServerMode.STARTUP);
-      server = new ServerThread();
-      server.start();
+      serverThread = new ServerThread();
+      serverThread.start();
     } finally {
       exit(token);
     }
@@ -147,10 +160,10 @@ public void stop() throws InterruptedException {
       // Can't stop something not running.
       if (isServerRunning()) {
         // Log the intent.
-        serverLogger.output("Crashing server process: " + server);
+        serverLogger.output("Crashing server process: " + serverThread);
         // Mark this as expected.
         this.setCrashExpected(true);
-        boolean result = server.shutdown();
+        boolean result = serverThread.shutdown();
         serverLogger.output("Server Stop Command Result: " + result);
       }
     } finally {
@@ -229,24 +242,112 @@ public void run() {
       }
     }
 
+    private Object getServerObject() {
+      return server;
+    }
+
     private synchronized boolean initializeServer(OutputStream out) throws Exception {
-        server = startIsolatedServer(serverWorkingDir, serverInstall, out, cmd, serverProperties);
-        running = server != null;
+      server = startIsolatedServer(serverWorkingDir, serverInstall, out, cmd, serverProperties);
+      running = server != null;
       return running;
     }
 
     public synchronized boolean shutdown() {
       if (running) {
         running = false;
-        String result = invokeOnServerMBean(server, "Server","halt",null);
-        serverLogger.output("stopping. " + result);
+        String result = invokeOnServerMBean(server, "Server","stopAndWait",null);
+        serverLogger.output("stopping. restart:" + result);
+        detectThreadLeaks();
         return !Boolean.parseBoolean(result);
       } else {
         return true;
       }
     }
   }
 
+  private void detectThreadLeaks() {
+    ThreadGroup grp = (ThreadGroup)invokeOnObject(serverThread.getServerObject(), "getServerThreadGroup");
+    List<Thread> threads = threads(grp);
+    if (!threads.isEmpty()) {
+      LOGGER.warn("Inline server threads are leaking:");
+      LOGGER.warn("server name:{} path:{}", serverName, serverWorkingDir);
+      for (Thread t : threads) {
+        LOGGER.warn("thread group: {}", climbThreadGroupChain(t.getThreadGroup()));
+        LOGGER.warn(getThreadDump(t));
+      }
+      grp.interrupt();
+      leakDetected = false;
+      boolean interrupted = false;
+      try {
+        for (Thread t : threads) {
+          try {
+            t.join(500L);
+            leakDetected = t.isAlive() | leakDetected;
+          } catch (InterruptedException ie) {
+            interrupted = true;
+          }
+        }
+      } finally {
+        if (interrupted) {
+          Thread.currentThread().interrupt();
+        }
+      }
+    }
+  }
+
+  private List<Thread> threads(ThreadGroup grp) {
+    if (grp != null) {
+      int ac = grp.activeCount();
+      Thread[] list = new Thread[ac];
+      grp.enumerate(list, true);
+      return Arrays.stream(list).filter(t -> t != null && t.isAlive()).collect(Collectors.toList());
+    } else {
+      return Collections.emptyList();
+    }
+  }
+
+  private List<ThreadGroup> climbThreadGroupChain(ThreadGroup tg) {
+    List<ThreadGroup> list = new ArrayList<>();
+    while (tg != null) {
+      list.add(tg);
+      tg = tg.getParent();
+    }
+    Collections.reverse(list);
+    return list;
+  }
+
+  private static String getThreadDump(Thread t) {
+    final StringBuilder sb = new StringBuilder(100 * 1024);
+    sb.append("name=");
+    sb.append(t.getName());
+    sb.append(" id=");
+    sb.append(t.getId());
+    sb.append('\n');
+
+    final StackTraceElement[] stea = t.getStackTrace();
+    for (StackTraceElement element : stea) {
+      sb.append("\tat ");
+      sb.append(element.toString());
+      sb.append('\n');
+    }
+    sb.append('\n');
+
+    return sb.toString();
+  }
+
+  @Override
+  public IGalvanServer newInstance() {
+    if (leakDetected) {
+      LOGGER.warn("Leak detected switching to process mode");
+      ServerProcess process = new ServerProcess(serverName, serverInstall, serverWorkingDir, heap,
+          debug, serverProperties, parentOutput, cmd);
+      process.installIntoStripe(stateInterlock, stateManager, this.serverLogger);
+      return process;
+    } else {
+      return this;
+    }
+  }
+
   private static Object startIsolatedServer(Path serverWorking, Path server, OutputStream out, String[] cmd, Properties serverProperties) {
     ClassLoader oldLoader = Thread.currentThread().getContextClassLoader();
     Path tc = server.resolve("tc.jar");

galvan/src/main/java/org/terracotta/testing/master/ServerInstance.java

@@ -56,10 +56,10 @@ public ServerInstance(String serverName) {
     this.serverName = serverName;
   }
 
-  public void installIntoStripe(StateInterlock stateInterlock, ITestStateManager stateManager, VerboseManager logging) {
+  public void installIntoStripe(StateInterlock stateInterlock, ITestStateManager stateManager, ContextualLogger logging) {
     this.stateInterlock = stateInterlock;
     this.stateManager = stateManager;
-    this.serverLogger = logging.createServerLogger();
+    this.serverLogger = logging;
     this.stateInterlock.registerNewServer(this);
   }
 
@@ -286,6 +286,11 @@ public synchronized void waitForState(ServerMode mode) throws InterruptedExcepti
     }
   }
 
+  @Override
+  public IGalvanServer newInstance() {
+    return this;
+  }
+
   @Override
   public String toString() {
     return "Server " + this.serverName + " (" + getCurrentState() + ")";

galvan/src/main/java/org/terracotta/testing/master/ServerProcessControl.java

@@ -101,7 +101,7 @@ public synchronized void startOneServer() throws GalvanFailureException {
     this.logger.output("<<< startOneServer");
   }
 
-  public synchronized void startServer() throws GalvanFailureException {
+  private synchronized void startServer() throws GalvanFailureException {
     IGalvanServer server = null;
     int tries = 1;
     while (null == (server = this.stateInterlock.getOneTerminatedServer())) {

galvan/src/main/java/org/terracotta/testing/master/StateInterlock.java

@@ -157,6 +157,13 @@ public IGalvanServer getOneTerminatedServer() throws GalvanFailureException {
         one = this.servers.stream().filter(s->s.getCurrentState() == ServerMode.TERMINATED).findAny().orElse(null);
       }
       this.logger.output("getOneTerminatedServer " + one);
+      if (one != null) {
+        IGalvanServer newOne = one.newInstance();
+        if (newOne != one) {
+          this.servers.remove(one);
+          one = newOne;
+        }
+      }
       return one;
   }
 
@@ -193,6 +200,7 @@ private boolean checkIfEmpty() {
         && this.runningClients.isEmpty();
   }
   // ----- CLEANUP-----
+  @Override
   public void forceShutdown() throws GalvanFailureException {
       this.logger.output("> forceShutdown");
       // Set the flag that we are shutting down.  That way, any servers which were concurrently coming online can be stopped when they check in.

galvan/src/main/java/org/terracotta/testing/master/StripeInstaller.java

@@ -43,7 +43,7 @@ public StripeInstaller(StateInterlock interlock, ITestStateManager stateManager,
   public void installNewServer(ServerInstance serverProcess) throws IOException {
     // Our implementation installs all servers before starting any (just an internal consistency check).
     Assert.assertFalse(this.isBuilt);
-    serverProcess.installIntoStripe(interlock, stateManager, stripeVerboseManager.createComponentManager("[" + serverProcess.serverName + "]"));
+    serverProcess.installIntoStripe(interlock, stateManager, stripeVerboseManager.createComponentManager("[" + serverProcess.serverName + "]").createServerLogger());
     serverProcesses.add(serverProcess);
   }