add new weighting scores to help balance threads

Author: Myron Scott

common/src/main/java/com/tc/net/core/CoreNIOServices.java

@@ -45,6 +45,7 @@
 import java.nio.channels.Selector;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
@@ -170,6 +171,10 @@ private void updateListenerString() {
   private synchronized String getListenerString() {
     return this.listenerString;
   }
+  
+  public long getCongestionScore() {
+    return readerComm.getCongestionScore() + writerComm.getCongestionScore();
+  }
 
   public long getTotalBytesRead() {
     return readerComm.getTotalBytesRead() + writerComm.getTotalBytesRead();
@@ -349,6 +354,8 @@ protected class CommThread extends Thread {
     private final String                        name;
     private long                    bytesMoved    = 0;
     private final COMM_THREAD_MODE              mode;
+    private long congestionScore = 0;
+    private long lastIdleNanos = System.nanoTime();
 
     public CommThread(COMM_THREAD_MODE mode) {
       name = commThreadName + (mode == COMM_THREAD_MODE.NIO_READER ? "_R" : "_W");
@@ -610,23 +617,21 @@ private void selectLoop() throws IOException {
 
       Selector localSelector = this.selector;
       Queue<Runnable> localSelectorTasks = this.selectorTasks;
-      int congestionDetected = 0;
 
       while (true) {
         final int numKeys;
         try {
-          if (isReader()) {
-            int localKeys = localSelector.selectNow();
-            if (localKeys == 0 && localSelectorTasks.isEmpty()) {
-              congestionDetected = 0;
+          int localKeys = localSelector.selectNow();
+          if (localKeys == 0 && localSelectorTasks.isEmpty()) {
+            congestionScore = 0;
+            lastIdleNanos = System.nanoTime();
+            if (!isStopRequested()) {
               localKeys = localSelector.select();
-            } else {
-              congestionDetected++;
             }
-            numKeys = localKeys;
           } else {
-            numKeys = localSelector.select();
+            congestionScore++;
           }
+          numKeys = localKeys;
         } catch (IOException ioe) {
           throw ioe;
         } catch (CancelledKeyException cke) {
@@ -663,7 +668,7 @@ private void selectLoop() throws IOException {
         Util.selfInterruptIfNeeded(isInterrupted);
 
         final Set<SelectionKey> selectedKeys = localSelector.selectedKeys();
-        if ((0 == numKeys) && (0 == selectedKeys.size())) {
+        if ((0 == numKeys) && (selectedKeys.isEmpty())) {
           continue;
         }
 
@@ -719,7 +724,9 @@ private void selectLoop() throws IOException {
             }
           }
         } // for
-        if (isReader() && congestionDetected > 10 && workerCommMgr != null && workerCommMgr.isUnbalanced()) {
+        if (isReader() && 
+                congestionScore > 100 && System.nanoTime() - lastIdleNanos > Duration.ofSeconds(2).toNanos() &&
+                workerCommMgr != null && workerCommMgr.isOverweight(getWeight())) {
           for (SelectionKey key : selectedKeys) {
             TCConnectionImpl connect = (TCConnectionImpl)key.attachment();
             connect.migrate();
@@ -783,6 +790,10 @@ private void doConnect(SelectionKey key) {
       }
     }
 
+    public long getCongestionScore() {
+      return congestionScore;
+    }
+    
     public long getTotalBytesRead() {
       return this.mode == COMM_THREAD_MODE.NIO_READER ? this.bytesMoved : 0;
     }

common/src/main/java/com/tc/net/core/TCWorkerCommManager.java

@@ -145,15 +145,12 @@ synchronized void waitDuringPause() throws IOException {
     }
   }
 
-  public boolean isUnbalanced() {
-    int min = Integer.MAX_VALUE;
-    int max = 0;
+  public boolean isOverweight(int weight) {
+    if (weight <= totalWorkerComm) return false;
     for (CoreNIOServices c : workerCommThreads) {
-      int w = c.getWeight();
-      min = Math.min(min, w);
-      max = Math.max(max, w);
+      if (c.getWeight() * 2 < weight && c.getCongestionScore() == 0) return true;
     }
-    return (max - min) > totalWorkerComm;
+    return false;
   }
 
   public synchronized void pause() {

common/src/test/java/com/tc/net/core/TCWorkerCommManagerTest.java

@@ -59,7 +59,7 @@ public TCWorkerCommManagerTest() {
   }
 
   private synchronized ClientMessageTransport createClient(String clientName) {
-    TCConnectionManager connection = new TCConnectionManagerImpl("Client-TestCommMgr", 0, new ClearTextSocketEndpointFactory());
+    TCConnectionManager connection = new TCConnectionManagerImpl("Client-TestCommMgr-" + clientName, 0, new ClearTextSocketEndpointFactory());
     clientConnectionMgrs.add(connection);
     CommunicationsManager commsMgr = new CommunicationsManagerImpl(new NullMessageMonitor(),
                                                                    new TransportNetworkStackHarnessFactory(),
@@ -131,6 +131,53 @@ public void testSimpleOpenAndClose() throws Exception {
     connMgr.shutdown();
   }
 
+  public void testOverweight() throws Exception {
+    TCConnectionManager connMgr = new TCConnectionManagerImpl("Target-Server-TestCommsMgr", 2, new ClearTextSocketEndpointFactory());
+    CommunicationsManager commsMgr = new CommunicationsManagerImpl(new NullMessageMonitor(),
+                                                                   new TransportNetworkStackHarnessFactory(),
+                                                                   connMgr,
+                                                                   new NullConnectionPolicy());
+    NetworkListener listener = commsMgr.createListener(new InetSocketAddress(0), (c)->true,
+                                                       new DefaultConnectionIdFactory(), (MessageTransport t)->true);
+    listener.start(Collections.<ConnectionID>emptySet());
+    int port = listener.getBindPort();
+    
+    ClientMessageTransport[] clients = new ClientMessageTransport[32];
+
+    for (int x=0;x<clients.length;x++) {
+      clients[x] = createBasicClient("client" + x);
+    }
+
+    InetSocketAddress serverAddress = InetSocketAddress.createUnresolved("localhost", port);
+
+    for (ClientMessageTransport t : clients) {
+      t.open(serverAddress);
+    }
+
+    waitForConnected(clients);
+    waitForTotalWeights(commsMgr,2, 32);
+    
+    for (int x=0;x<clients.length;x++) {
+      if (x%2 == 0) {
+        clients[x].close();
+      }
+    }
+        
+    waitForTotalWeights(commsMgr,2, 16);
+
+    TCCommImpl comm = (TCCommImpl)connMgr.getTcComm();
+    Assert.assertEquals(0, comm.getWeightForWorkerComm(0));
+    Assert.assertEquals(16, comm.getWeightForWorkerComm(1));
+    
+    TCConnection[] all = connMgr.getAllConnections();
+    for (int x=0;x<all.length/2;x++) {
+      ((TCConnectionImpl)all[x]).migrate();
+    }
+    waitForWeight(commsMgr,0,8);
+    Assert.assertEquals(comm.getWeightForWorkerComm(0), 8);
+    Assert.assertEquals(comm.getWeightForWorkerComm(1), 8);
+  }
+
   public void testReaderandWriterCommThread() throws Exception {
     // comms manager with 4 worker comms
     logger.debug("Running target test");
@@ -367,8 +414,10 @@ private static void waitForWeight(final CommunicationsManager communicationsMana
     CallableWaiter.waitOnCallable(new Callable<Boolean>() {
       @Override
       public Boolean call() throws Exception {
-        return ((TCCommImpl)communicationsManager.getConnectionManager()
-            .getTcComm()).getWeightForWorkerComm(commId) == weight;
+        int w = ((TCCommImpl)communicationsManager.getConnectionManager()
+            .getTcComm()).getWeightForWorkerComm(commId);
+        System.out.println("waiting for id:" + commId + " weight " + w + " expected " + weight);
+        return w == weight;
       }
     });
   }

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

@@ -172,7 +172,7 @@ public void testHealthCheckThreadDies() throws Exception {
     CLUSTER.getClusterControl().startAllServers();
     CLUSTER.getClusterControl().waitForActive();
 
-    String connectionName = "LEAKTESTCLIENT";
+    String connectionName = "LEAKTESTCLIENT2";
     Properties props = new Properties();
     props.setProperty(ConnectionPropertyNames.CONNECTION_NAME, connectionName);
     Connection leak = ConnectionFactory.connect(CLUSTER.getConnectionURI(), props);