Move to a single health-check timer per classloader. Add healthchecker to diagnostic connections

Author: Myron Scott

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

@@ -73,6 +73,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 
 /**
  * Communications manager for setting up listeners and creating client connections
@@ -96,7 +97,7 @@ public class CommunicationsManagerImpl implements CommunicationsManager {
   protected final ConcurrentHashMap<TCMessageType, Class<? extends TCAction>> messageTypeClassMapping   = new ConcurrentHashMap<TCMessageType, Class<? extends TCAction>>();
   protected final ConcurrentHashMap<TCMessageType, GeneratedMessageFactory>    messageTypeFactoryMapping = new ConcurrentHashMap<TCMessageType, GeneratedMessageFactory>();
 
-  private ConnectionHealthChecker                                              connectionHealthChecker;
+  private final ConnectionHealthChecker                                              connectionHealthChecker;
   private ServerID                                                             serverID                  = ServerID.NULL_ID;
   private int                                                                  callbackPort              = TransportHandshakeMessage.NO_CALLBACK_PORT;
   private final TransportHandshakeErrorHandler                                 handshakeErrHandler;
@@ -177,15 +178,22 @@ public CommunicationsManagerImpl(MessageMonitor monitor, TCMessageRouter message
     this.connectionManager = connMgr;
 
     Assert.eval(healthCheckerConfig != null);
-    if (healthCheckerConfig.isHealthCheckerEnabled()) {
+    connectionHealthChecker = healthCheckerConfig.isHealthCheckerEnabled() ?
+            new ConnectionHealthCheckerImpl(healthCheckerConfig, connectionManager, createReferenceCheck()) :
+            new ConnectionHealthCheckerEchoImpl();
+  }
+  
+  private Supplier<Boolean> createReferenceCheck() {
 // reference this manager.  If it not reachable, all connections and threads associated need to be cleaned up.
-      ReferenceQueue<Object> gc = new ReferenceQueue<>();
-      PhantomReference<Object> ref = new PhantomReference<>(this, gc);
-      connectionHealthChecker = new ConnectionHealthCheckerImpl(healthCheckerConfig, connectionManager, ()->gc.poll() != ref);
-    } else {
-      connectionHealthChecker = new ConnectionHealthCheckerEchoImpl();
-    }
-    connectionHealthChecker.start();
+    ReferenceQueue<Object> gc = new ReferenceQueue<>();
+    PhantomReference<Object> ref = new PhantomReference<>(this, gc);
+    SetOnceFlag gcd = new SetOnceFlag();
+    return ()-> {
+      if (gc.poll() == ref) {
+        gcd.set();
+      }
+      return !gcd.isSet();
+    };
   }
 
   @Override
@@ -392,12 +400,6 @@ public ConnectionHealthChecker getConnHealthChecker() {
     return this.connectionHealthChecker;
   }
 
-  public void setConnHealthChecker(ConnectionHealthChecker checker) {
-    this.connectionHealthChecker.stop();
-    this.connectionHealthChecker = checker;
-    this.connectionHealthChecker.start();
-  }
-
   public TCMessageRouter getMessageRouter() {
     return this.messageRouter;
   }

common/src/main/java/com/tc/net/protocol/transport/ConnectionHealthCheckerImpl.java

@@ -18,20 +18,22 @@
  */
 package com.tc.net.protocol.transport;
 
+import com.tc.lang.TCThreadGroup;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.tc.net.core.TCConnection;
 import com.tc.net.core.TCConnectionManager;
 import com.tc.util.Assert;
-import com.tc.util.concurrent.SetOnceFlag;
 import java.net.InetSocketAddress;
 
 import java.util.Iterator;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 
@@ -44,20 +46,35 @@
 public class ConnectionHealthCheckerImpl implements ConnectionHealthChecker {
 
   private final Logger logger;
-  private final Timer                           monitorThread;
+  private static final ScheduledExecutorService monitorThread = createHealthCheckExecutor();
   private final HealthCheckerMonitorThreadEngine monitorThreadEngine;
 
-  private final SetOnceFlag                      shutdown = new SetOnceFlag();
-  private final SetOnceFlag                      started  = new SetOnceFlag();
+  private Future<?>                      task;
 
   public ConnectionHealthCheckerImpl(HealthCheckerConfig healthCheckerConfig, TCConnectionManager connManager, Supplier<Boolean> reachable) {
     Assert.assertNotNull(healthCheckerConfig);
     Assert.eval(healthCheckerConfig.isHealthCheckerEnabled());
     logger = LoggerFactory.getLogger(ConnectionHealthCheckerImpl.class.getName() + ": "
                                      + healthCheckerConfig.getHealthCheckerName());
-    monitorThread = new Timer(healthCheckerConfig.getHealthCheckerName() + " - HealthCheck-Timer", true);
     monitorThreadEngine = getHealthMonitorThreadEngine(healthCheckerConfig, connManager, reachable, logger);
   }
+  
+  private static ScheduledExecutorService createHealthCheckExecutor() {
+    ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1, (r) -> {
+      ThreadGroup grp = Thread.currentThread().getThreadGroup();
+      String id = "";
+      while (grp instanceof TCThreadGroup) {
+        id = "-" + grp.getName();
+        grp = grp.getParent();
+      }
+      Thread t = new Thread(grp, r, "HealthCheck" + id);
+      t.setDaemon(true);
+      return t;
+    });
+    exec.setKeepAliveTime(5, TimeUnit.SECONDS);
+    exec.allowCoreThreadTimeOut(true);
+    return exec;
+  }
 
   private HealthCheckerMonitorThreadEngine getHealthMonitorThreadEngine(HealthCheckerConfig config,
                                                                           TCConnectionManager connectionManager,
@@ -69,17 +86,18 @@ private HealthCheckerMonitorThreadEngine getHealthMonitorThreadEngine(HealthChec
       } else {
         stop();
         connectionManager.asynchCloseAllConnections();
+        connectionManager.closeAllListeners();
         connectionManager.shutdown();
         return false;
       }
     }, logger);
   }
 
   @Override
-  public void start() {
-    if (started.attemptSet() && !shutdown.isSet()) {
+  public synchronized void start() {
+    if (task == null) {
       try {
-        monitorThread.scheduleAtFixedRate(monitorThreadEngine, 0L, monitorThreadEngine.pingInterval);
+        task = monitorThread.scheduleAtFixedRate(monitorThreadEngine, 0L, monitorThreadEngine.pingInterval, TimeUnit.MILLISECONDS);
       } catch (IllegalStateException state) {
         logger.warn("HealthChecker cannot start");
         return;
@@ -91,22 +109,15 @@ public void start() {
   }
 
   @Override
-  public void stop() {
-    if (shutdown.attemptSet()) {
-      monitorThreadEngine.stop();
-      monitorThread.cancel();
-// don't bother to join the monitorThread here.  shutdown should take care of all the
-// threads in the thread group
+  public synchronized void stop() {
+    if (task != null && !task.isCancelled()) {
+      task.cancel(true);
       logger.debug("HealthChecker STOP requested");
     } else {
       logger.warn("HealthChecker STOP already requested");
     }
   }
 
-  public boolean isRunning() {
-    return started.isSet() && !shutdown.isSet();
-  }
-
   @Override
   public void notifyTransportClosed(MessageTransport transport) {
     // HealthChecker Ping Thread can anyway determine this in the next probe interval thru mtb.isConnected and remove it
@@ -129,6 +140,7 @@ public void notifyTransportConnectAttempt(MessageTransport transport) {
 
   @Override
   public void notifyTransportConnected(MessageTransport transport) {
+    start();
     monitorThreadEngine.addConnection(transport);
   }
 
@@ -153,7 +165,7 @@ public void notifyTransportReconnectionRejected(MessageTransport transport) {
     // NOP
   }
 
-  static class HealthCheckerMonitorThreadEngine extends TimerTask {
+  static class HealthCheckerMonitorThreadEngine implements Runnable {
     private final ConcurrentMap<ConnectionID, MessageTransportBase> connectionMap =
         new ConcurrentHashMap<>();
     private final long                pingIdleTime;
@@ -204,10 +216,6 @@ protected ConnectionHealthCheckerContext getHealthCheckerContext(MessageTranspor
       return new ConnectionHealthCheckerContextImpl(transport, conf, connManager);
     }
 
-    public void stop() {
-      this.cancel();
-    }
-
     @Override
     public void run() {
       // same interval for all connections

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

@@ -19,6 +19,7 @@
 package org.terracotta.testing.rules;
 
 
+import java.net.InetSocketAddress;
 import java.util.Properties;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -31,6 +32,8 @@
 import org.terracotta.connection.ConnectionException;
 import org.terracotta.connection.ConnectionFactory;
 import org.terracotta.connection.ConnectionPropertyNames;
+import org.terracotta.connection.Diagnostics;
+import org.terracotta.connection.DiagnosticsFactory;
 
 
 /**
@@ -147,8 +150,54 @@ public void testSEDADiesWithNoRef() throws Exception {
       System.gc();
       Thread.sleep(1_000);
     }
+    
+    String[] hp = CLUSTER.getClusterHostPorts();
+    String[] server = hp[0].split(":");
+    Diagnostics diag = DiagnosticsFactory.connect(InetSocketAddress.createUnresolved(server[0], Integer.parseInt(server[1])), props);
+
+    assertNotNull(diag);
+    assertNotNull(lookForThreadWithName(connectionName));
+    
+    diag = null;
+    
+    while (lookForThreadWithName(connectionName) != null) {
+      LOGGER.info(cid);
+      System.gc();
+      Thread.sleep(1_000);
+    }    
   }
+    
+  @Test
+  public void testHealthCheckThreadDies() throws Exception {
+    CLUSTER.getClusterControl().startAllServers();
+    CLUSTER.getClusterControl().waitForActive();
+
+    String connectionName = "LEAKTESTCLIENT";
+    Properties props = new Properties();
+    props.setProperty(ConnectionPropertyNames.CONNECTION_NAME, connectionName);
+    Connection leak = ConnectionFactory.connect(CLUSTER.getConnectionURI(), props);
+    String cid = leak.toString();
 
+    assertNotNull(leak);
+    assertNotNull(lookForThreadWithName(connectionName));
+
+    leak = null;
+    
+    while (lookForThreadWithName(connectionName) != null) {
+      LOGGER.info(cid);
+      System.gc();
+      Thread.sleep(1_000);
+    }
+    
+    CLUSTER.getClusterControl().terminateAllServers();
+    
+    while (lookForThreadWithName("HealthCheck") != null) {
+      LOGGER.info(cid);
+      System.gc();
+      Thread.sleep(1_000);
+    }
+  }
+  
   private static Thread lookForThreadWithName(String name) {
     Thread[] list = new Thread[Thread.activeCount()];
     Thread.enumerate(list);

tc-client/src/main/java/com/terracotta/diagnostic/DiagnosticClientBuilder.java

@@ -19,40 +19,22 @@
 package com.terracotta.diagnostic;
 
 
-import com.tc.async.api.StageManager;
 import com.tc.net.basic.BasicConnectionManager;
 import com.tc.net.core.BufferManagerFactory;
 import com.tc.net.core.TCConnectionManager;
-import com.tc.net.protocol.NetworkStackHarnessFactory;
 import com.tc.net.protocol.tcm.ClientMessageChannel;
-import com.tc.net.protocol.tcm.CommunicationsManager;
-import com.tc.net.protocol.tcm.MessageMonitor;
-import com.tc.net.protocol.tcm.TCMessageRouter;
-import com.tc.net.protocol.tcm.TCMessageType;
-import com.tc.net.protocol.transport.ConnectionPolicy;
-import com.tc.net.protocol.transport.DisabledHealthCheckerConfigImpl;
-import com.tc.net.protocol.transport.HealthCheckerConfig;
-import com.tc.net.protocol.transport.ReconnectionRejectedHandler;
 import com.tc.object.ClientEntityManager;
 import com.tc.object.StandardClientBuilder;
 import com.tc.net.core.ProductID;
-import java.util.Map;
 
 import java.util.Properties;
-import com.tc.net.protocol.tcm.TCAction;
 
 
 public class DiagnosticClientBuilder extends StandardClientBuilder {
 
   public DiagnosticClientBuilder(Properties connectionProperties, BufferManagerFactory buffers) {
     super(connectionProperties, buffers);
   }
-
-  @Override
-  public CommunicationsManager createCommunicationsManager(MessageMonitor monitor, TCMessageRouter messageRouter, NetworkStackHarnessFactory stackHarnessFactory, ConnectionPolicy connectionPolicy, 
-          TCConnectionManager connections, HealthCheckerConfig aConfig, Map<TCMessageType, Class<? extends TCAction>> messageTypeClassMapping, ReconnectionRejectedHandler reconnectionRejectedHandler) {
-    return super.createCommunicationsManager(monitor, messageRouter, stackHarnessFactory, connectionPolicy, connections, new DisabledHealthCheckerConfigImpl(), messageTypeClassMapping, reconnectionRejectedHandler); 
-  }
 
   @Override
   public ClientEntityManager createClientEntityManager(ClientMessageChannel channel) {