Merge pull request #1284 from myronkscott/reference_cleanup

Reference tracking reduction

Author: chrisdennis

common/src/main/java/com/tc/async/api/SEDA.java

@@ -33,12 +33,12 @@ public class SEDA {
 
   public SEDA(TCThreadGroup threadGroup) {
     this.threadGroup = threadGroup;
-    this.stageManager = new StageManagerImpl(threadGroup, new QueueFactory(), new StageListener() {
-      @Override
-      public void stageStalled(String name, long delay, int queueDepth) {
-        stageWarning(new StallWarning(name, delay, queueDepth));
-      }
-    });
+    this.stageManager = new StageManagerImpl(threadGroup, new QueueFactory());
+  }
+  
+  public SEDA(TCThreadGroup threadGroup, StageListener listener) {
+    this.threadGroup = threadGroup;
+    this.stageManager = new StageManagerImpl(threadGroup, new QueueFactory(), listener);
   }
 
   public StageManager getStageManager() {
@@ -48,25 +48,4 @@ public StageManager getStageManager() {
   protected TCThreadGroup getThreadGroup() {
     return this.threadGroup;
   }
-  
-  public void stageWarning(Object description) {
-    
-  }
-  
-  private static class StallWarning {
-    private final String name;
-    private final long delay;
-    private final int depth;
-
-    public StallWarning(String name, long delay, int depth) {
-      this.name = name;
-      this.delay = delay;
-      this.depth = depth;
-    }
-
-    @Override
-    public String toString() {
-      return "StallWarning{" + "name=" + name + ", delay=" + delay + ", depth=" + depth + '}';
-    }
-  }
 }

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

@@ -60,6 +60,8 @@
 import com.tc.util.concurrent.SetOnceFlag;
 
 import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
 import java.net.InetSocketAddress;
 import java.util.Collections;
 import java.util.HashSet;
@@ -176,7 +178,10 @@ public CommunicationsManagerImpl(MessageMonitor monitor, TCMessageRouter message
 
     Assert.eval(healthCheckerConfig != null);
     if (healthCheckerConfig.isHealthCheckerEnabled()) {
-      connectionHealthChecker = new ConnectionHealthCheckerImpl(healthCheckerConfig, connectionManager);
+// 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();
     }

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

@@ -33,6 +33,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
 
 /**
  * The Engine which does the peer health checking work. Based on the config passed, it probes the peer once in specified
@@ -48,20 +49,30 @@ public class ConnectionHealthCheckerImpl implements ConnectionHealthChecker {
 
   private final SetOnceFlag                      shutdown = new SetOnceFlag();
   private final SetOnceFlag                      started  = new SetOnceFlag();
-
-  public ConnectionHealthCheckerImpl(HealthCheckerConfig healthCheckerConfig, TCConnectionManager connManager) {
+  
+  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, logger);
+    monitorThreadEngine = getHealthMonitorThreadEngine(healthCheckerConfig, connManager, reachable, logger);
   }
 
-  protected HealthCheckerMonitorThreadEngine getHealthMonitorThreadEngine(HealthCheckerConfig config,
+  private HealthCheckerMonitorThreadEngine getHealthMonitorThreadEngine(HealthCheckerConfig config,
                                                                           TCConnectionManager connectionManager,
-                                                                          Logger loger) {
-    return new HealthCheckerMonitorThreadEngine(config, connectionManager, loger);
+                                                                          Supplier<Boolean> reachable,
+                                                                          Logger logger) {
+    return new HealthCheckerMonitorThreadEngine(config, connectionManager, ()-> {
+      if (reachable.get()) {
+        return true;
+      } else {
+        stop();
+        connectionManager.asynchCloseAllConnections();
+        connectionManager.shutdown();
+        return false;
+      }
+    }, logger);
   }
 
   @Override
@@ -149,19 +160,20 @@ static class HealthCheckerMonitorThreadEngine extends TimerTask {
     private final long                pingInterval;
     private final int                 pingProbes;
     private final long                checkTimeInterval;
-    private final SetOnceFlag         stop          = new SetOnceFlag();
     private final HealthCheckerConfig config;
     private final Logger logger;
     private final TCConnectionManager connectionManager;
+    private final Supplier<Boolean>   reachable;
     private final AtomicLong          lastCheckTime = new AtomicLong(System.currentTimeMillis());
 
     public HealthCheckerMonitorThreadEngine(HealthCheckerConfig healthCheckerConfig,
-                                            TCConnectionManager connectionManager, Logger logger) {
+                                            TCConnectionManager connectionManager, Supplier<Boolean> reachable, Logger logger) {
       this.pingIdleTime = healthCheckerConfig.getPingIdleTimeMillis();
       this.pingInterval = healthCheckerConfig.getPingIntervalMillis();
       this.pingProbes = healthCheckerConfig.getPingProbes();
       this.checkTimeInterval = healthCheckerConfig.getCheckTimeInterval();
       this.connectionManager = connectionManager;
+      this.reachable = reachable;
       this.config = healthCheckerConfig;
 
       Assert.assertNotNull(logger);
@@ -193,7 +205,6 @@ protected ConnectionHealthCheckerContext getHealthCheckerContext(MessageTranspor
     }
 
     public void stop() {
-      stop.attemptSet();
       this.cancel();
     }
 
@@ -202,42 +213,44 @@ public void run() {
       // same interval for all connections
       final boolean canCheckTime = canCheckTime();
 
-      Iterator<MessageTransportBase> connectionIterator = connectionMap.values().iterator();
-      while (connectionIterator.hasNext()) {
-        MessageTransportBase mtb = connectionIterator.next();
-
-        TCConnection conn = mtb.getConnection();
-        if (conn == null || !mtb.isConnected()) {
-          logger.info("[" + (conn == null ? null : conn.getRemoteAddress().toString())
-                      + "] is not connected. Health Monitoring for this node is now disabled.");
-          connectionIterator.remove();
-          continue;
-        }
-
-        if (mtb.getReceiveLayer() == null) {
-          logger.info("[" + (conn == null ? null : conn.getRemoteAddress().toString())
-                      + "] is no longer referenced.  Closing the connection");
-          mtb.disconnect();
-          connectionIterator.remove();
-          continue;
-        }
+      if (reachable.get()) {
+        Iterator<MessageTransportBase> connectionIterator = connectionMap.values().iterator();
+        while (connectionIterator.hasNext()) {
+          MessageTransportBase mtb = connectionIterator.next();
 
-        ConnectionHealthCheckerContext connContext = mtb.getHealthCheckerContext();
-        if ((conn.getIdleReceiveTime() >= this.pingIdleTime)) {
+          TCConnection conn = mtb.getConnection();
+          if (conn == null || !mtb.isConnected()) {
+            logger.info("[" + (conn == null ? null : conn.getRemoteAddress().toString())
+                        + "] is not connected. Health Monitoring for this node is now disabled.");
+            connectionIterator.remove();
+            continue;
+          }
 
-          if (!connContext.probeIfAlive()) {
-            // Connection is dead. Disconnect the transport.
-            logger.error("Declared connection dead " + mtb.getConnectionID() + " idle time "
-                         + conn.getIdleReceiveTime() + "ms");
+          if (mtb.getReceiveLayer() == null) {
+            logger.info("[" + (conn == null ? null : conn.getRemoteAddress().toString())
+                        + "] is no longer referenced.  Closing the connection");
             mtb.disconnect();
             connectionIterator.remove();
+            continue;
+          }
+
+          ConnectionHealthCheckerContext connContext = mtb.getHealthCheckerContext();
+          if ((conn.getIdleReceiveTime() >= this.pingIdleTime)) {
+
+            if (!connContext.probeIfAlive()) {
+              // Connection is dead. Disconnect the transport.
+              logger.error("Declared connection dead " + mtb.getConnectionID() + " idle time "
+                           + conn.getIdleReceiveTime() + "ms");
+              mtb.disconnect();
+              connectionIterator.remove();
+            }
+          } else {
+            connContext.refresh();
+          }
+          // is there any significant time difference between hosts ?
+          if (canCheckTime) {
+            connContext.checkTime();
           }
-        } else {
-          connContext.refresh();
-        }
-        // is there any significant time difference between hosts ?
-        if (canCheckTime) {
-          connContext.checkTime();
         }
       }
 

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

@@ -39,19 +39,21 @@ public class ConnectionWatcher implements MessageTransportListener {
    * Listens to events from a MessageTransport, acts on them, and passes events through to target
    */
   public ConnectionWatcher(ClientMessageChannel target, ClientConnectionEstablisher cce) {
+  // this the channel is no longer reachable, make sure all the the connection threads are cleaned up
     this.targetHolder = new WeakReference<>(target, stopQueue);
     this.cce = cce;
     this.connection = target.getConnectionID();
   }
 
   private boolean checkForStop() {
     Reference<? extends ClientMessageChannel> target = stopQueue.poll();
-    if (target != null) {
+    while (target != null) {
       if (target == targetHolder) {
         stopped.set();
         LOGGER.warn("unreferenced connection left open {} {}", targetHolder.get(), connection);
         cce.shutdown();
       }
+      target = stopQueue.poll();
     }
     return stopped.isSet();
   }

common/src/test/java/com/tc/net/protocol/transport/HealthCheckerMonitorThreadEngineTest.java

@@ -46,7 +46,7 @@ public void testAllowCheckTimeIfEnabledInConfig() {
     // ignore interval
     props.setProperty("checkTime.interval", "-1");
     HealthCheckerConfigImpl config = new HealthCheckerConfigImpl(props, "test-config");
-    final HealthCheckerMonitorThreadEngine engine = new HealthCheckerMonitorThreadEngine(config, null, logger);
+    final HealthCheckerMonitorThreadEngine engine = new HealthCheckerMonitorThreadEngine(config, null, ()->Boolean.TRUE, logger);
 
     assertTrue(engine.canCheckTime());
   }
@@ -57,7 +57,7 @@ public void testDisallowCheckTimeIfDisabledInConfig() {
     // disable time checking
     props.setProperty("checkTime.enabled", "false");
     HealthCheckerConfigImpl config = new HealthCheckerConfigImpl(props, "test-config");
-    final HealthCheckerMonitorThreadEngine engine = new HealthCheckerMonitorThreadEngine(config, null, logger);
+    final HealthCheckerMonitorThreadEngine engine = new HealthCheckerMonitorThreadEngine(config, null, ()->Boolean.TRUE, logger);
 
     assertFalse(engine.canCheckTime());
   }
@@ -68,7 +68,7 @@ public void testDisallowCheckTimeIfIntervalNotExceeded() {
     // set short interval
     props.setProperty("checkTime.interval", "900000");
     HealthCheckerConfigImpl config = new HealthCheckerConfigImpl(props, "test-config");
-    final HealthCheckerMonitorThreadEngine engine = new HealthCheckerMonitorThreadEngine(config, null, logger);
+    final HealthCheckerMonitorThreadEngine engine = new HealthCheckerMonitorThreadEngine(config, null, ()->Boolean.TRUE, logger);
 
     assertFalse(engine.canCheckTime());
   }

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

@@ -19,12 +19,18 @@
 package org.terracotta.testing.rules;
 
 
+import java.util.Properties;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import org.junit.ClassRule;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.terracotta.connection.Connection;
 import org.terracotta.connection.ConnectionException;
+import org.terracotta.connection.ConnectionFactory;
+import org.terracotta.connection.ConnectionPropertyNames;
 
 
 /**
@@ -35,6 +41,8 @@
 // XXX: Currently ignored since this test depends on restartability of the server, which now requires a persistence service
 //  to be plugged in (and there isn't one available, in open source).
 public class ClientLeakIT {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClientLeakIT.class);
+  
   @ClassRule
   public static final Cluster CLUSTER = BasicExternalClusterBuilder.newCluster(1).build(); //logConfigExtensionResourceName("custom-logback-ext.xml").build();
 
@@ -117,26 +125,46 @@ public void testConnectionMakerDiesWithNoRef() throws Exception {
       assertFalse(maker.isAlive());
     }
   }
-  
-  private static Thread lookForConnectionMaker() {
+    
+  @Test
+  public void testSEDADiesWithNoRef() 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);
+    }
+  }
+
+  private static Thread lookForThreadWithName(String name) {
     Thread[] list = new Thread[Thread.activeCount()];
     Thread.enumerate(list);
     for (Thread t : list) {
-      if (t != null && t.getName().startsWith("Connection Maker")) {
+      if (t != null && t.getName().startsWith(name)) {
         return t;
       }
     }
     return null;
+  } 
+  
+  private static Thread lookForConnectionMaker() {
+    return lookForThreadWithName("Connection Maker");
   }  
 
   private static Thread lookForConnectionEstablisher() {
-    Thread[] list = new Thread[Thread.activeCount()];
-    Thread.enumerate(list);
-    for (Thread t : list) {
-      if (t != null && t.getName().startsWith("ConnectionEstablisher")) {
-        return t;
-      }
-    }
-    return null;
+    return lookForThreadWithName("ConnectionEstablisher");
   }
 }

tc-client/src/main/java/com/tc/object/ClientBuilder.java

@@ -61,7 +61,7 @@ ClientHandshakeManager createClientHandshakeManager(Logger logger,
                                                       String clientRevision,
                                                       ClientEntityManager entity);
 
-  ClientEntityManager createClientEntityManager(ClientMessageChannel channel, StageManager stages);
+  ClientEntityManager createClientEntityManager(ClientMessageChannel channel);
 
   void setClientConnectionErrorListener(ClientConnectionErrorListener listener);