ENHANCE: Create hashUpdateThreadPool to decrease IO thread overhead.

Author: brido4125

src/main/java/net/spy/memcached/CacheListUpdateTask.java

@@ -0,0 +1,30 @@
+package net.spy.memcached;
+
+
+import java.util.List;
+import java.util.concurrent.Callable;
+
+public class CacheListUpdateTask implements Callable<Boolean> {
+
+  private final MemcachedConnection memcachedConnection;
+  private final List<MemcachedNode> attachNodes;
+  private final List<MemcachedNode> removeNodes;
+
+  public CacheListUpdateTask(MemcachedConnection memcachedConnection,
+                             List<MemcachedNode> attachNodes,
+                             List<MemcachedNode> removeNodes) {
+
+    this.memcachedConnection = memcachedConnection;
+    this.attachNodes = attachNodes;
+    this.removeNodes = removeNodes;
+  }
+
+  @Override
+  public Boolean call() throws Exception {
+    memcachedConnection.getLocator().update(attachNodes, removeNodes);
+
+    // Remove the unavailable nodes.
+    memcachedConnection.handleNodesToRemove(removeNodes);
+    return true;
+  }
+}

src/main/java/net/spy/memcached/HashRingUpdateService.java

@@ -0,0 +1,21 @@
+package net.spy.memcached;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+public class HashRingUpdateService {
+  private final ExecutorService pool;
+
+  public HashRingUpdateService() {
+    pool = Executors.newSingleThreadExecutor();
+  }
+
+  public Future<Boolean> updateHashes(CacheListUpdateTask task) {
+    return pool.submit(task);
+  }
+
+  public void shutdown() {
+    pool.shutdown();
+  }
+}

src/main/java/net/spy/memcached/MemcachedConnection.java

@@ -45,6 +45,7 @@
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -115,6 +116,9 @@ public final class MemcachedConnection extends SpyObject {
   private final DelayedSwitchoverGroups delayedSwitchoverGroups =
       new DelayedSwitchoverGroups(DELAYED_SWITCHOVER_TIMEOUT_MILLISECONDS);
   /* ENABLE_REPLICATION end */
+  private final HashRingUpdateService hashUpdateService = new HashRingUpdateService();
+
+  private Future<Boolean> hashUpdateResult;
 
   /**
    * Construct a memcached connection.
@@ -313,7 +317,7 @@ public void handleIO() throws IOException {
     }
   }
 
-  private void handleNodesToRemove(final List<MemcachedNode> nodesToRemove) {
+  void handleNodesToRemove(final List<MemcachedNode> nodesToRemove) {
     for (MemcachedNode node : nodesToRemove) {
       getLogger().info("old memcached node removed %s", node);
       reconnectQueue.remove(node);
@@ -340,10 +344,9 @@ private void handleNodesToRemove(final List<MemcachedNode> nodesToRemove) {
     }
   }
 
-  private void updateConnections(List<InetSocketAddress> addrs) throws IOException {
-    List<MemcachedNode> attachNodes = new ArrayList<MemcachedNode>();
-    List<MemcachedNode> removeNodes = new ArrayList<MemcachedNode>();
-
+  private void getUpdateNodes(List<InetSocketAddress> addrs,
+                              List<MemcachedNode> attachNodes,
+                              List<MemcachedNode> removeNodes) throws IOException {
     for (MemcachedNode node : locator.getAll()) {
       if (addrs.contains(node.getSocketAddress())) {
         addrs.remove(node.getSocketAddress());
@@ -356,12 +359,6 @@ private void updateConnections(List<InetSocketAddress> addrs) throws IOException
     for (SocketAddress sa : addrs) {
       attachNodes.add(attachMemcachedNode(sa));
     }
-
-    // Update the hash.
-    locator.update(attachNodes, removeNodes);
-
-    // Remove the unavailable nodes.
-    handleNodesToRemove(removeNodes);
   }
 
   /* ENABLE_REPLICATION if */
@@ -704,7 +701,12 @@ void handleCacheNodesChange() throws IOException {
         return;
       }
       /* ENABLE_REPLICATION end */
-      updateConnections(AddrUtil.getAddresses(cacheList));
+      List<MemcachedNode> attachNodes = new ArrayList<MemcachedNode>();
+      List<MemcachedNode> removeNodes = new ArrayList<MemcachedNode>();
+      getUpdateNodes(AddrUtil.getAddresses(cacheList), attachNodes, removeNodes);
+      // Update the hash.
+      CacheListUpdateTask task = new CacheListUpdateTask(this, attachNodes, removeNodes);
+      hashUpdateResult = hashUpdateService.updateHashes(task);
     }
     /* ENABLE_MIGRATION if */
     if (arcusMigrEnabled && alterList != null) {
@@ -725,6 +727,16 @@ void handleCacheNodesChange() throws IOException {
     /* ENABLE_MIGRATION end */
   }
 
+  // Called By MemcachedConnectionTest.
+  boolean getHashUpdateResult() {
+    try {
+      hashUpdateResult.get();
+    } catch (Exception e) {
+      return false;
+    }
+    return true;
+  }
+
   // Called by CacheManger to add the memcached server group.
   public void setCacheNodesChange(String addrs) {
     String old = cacheNodesChange.getAndSet(addrs);
@@ -1533,6 +1545,7 @@ public void shutdown() throws IOException {
       }
     }
     selector.close();
+    hashUpdateService.shutdown();
     getLogger().debug("Shut down selector %s", selector);
   }
 

src/test/java/net/spy/memcached/MemcachedConnectionTest.java

@@ -72,7 +72,8 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    assertTrue(conn.getHashUpdateResult());
+    assertEquals(1, locator.getAll().size());
 
     // when
     conn.setCacheNodesChange("0.0.0.0:11211,0.0.0.0:11212,0.0.0.0:11213");
@@ -81,7 +82,8 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(3 == locator.getAll().size());
+    assertTrue(conn.getHashUpdateResult());
+    assertEquals(3, locator.getAll().size());
 
     // when
     conn.setCacheNodesChange("0.0.0.0:11212");
@@ -90,7 +92,8 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    assertTrue(conn.getHashUpdateResult());
+    assertEquals(1, locator.getAll().size());
   }
 
   public void testNodesChangeQueue_empty() throws Exception {
@@ -101,7 +104,7 @@ public void testNodesChangeQueue_empty() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(0 == locator.getAll().size());
+    assertEquals(0, locator.getAll().size());
   }
 
   public void testNodesChangeQueue_invalid_addr() {
@@ -128,7 +131,8 @@ public void testNodesChangeQueue_redundent() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(2 == locator.getAll().size());
+    assertTrue(conn.getHashUpdateResult());
+    assertEquals(2, locator.getAll().size());
   }
 
   public void testNodesChangeQueue_twice() throws Exception {
@@ -140,7 +144,8 @@ public void testNodesChangeQueue_twice() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    assertTrue(conn.getHashUpdateResult());
+    assertEquals(1, locator.getAll().size());
   }
 
   public void testAddOperations() throws Exception {