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,26 @@
+package net.spy.memcached;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class HashRingUpdateService {
+  private final ExecutorService pool;
+
+  public HashRingUpdateService() {
+    pool = Executors.newSingleThreadExecutor();
+  }
+
+  public Boolean updateHashes(CacheListUpdateTask task) {
+    Boolean res = null;
+    try {
+      res = pool.submit(task).get();
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+    return res;
+  }
+
+  public void shutdown() {
+    pool.shutdown();
+  }
+}

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

@@ -115,6 +115,7 @@ 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();
 
   /**
    * Construct a memcached connection.
@@ -313,7 +314,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);
@@ -358,10 +359,9 @@ private void updateConnections(List<InetSocketAddress> addrs) throws IOException
     }
 
     // Update the hash.
-    locator.update(attachNodes, removeNodes);
-
-    // Remove the unavailable nodes.
-    handleNodesToRemove(removeNodes);
+    hashUpdateService.updateHashes(new CacheListUpdateTask(
+            this, attachNodes, removeNodes
+    ));
   }
 
   /* ENABLE_REPLICATION if */
@@ -1533,6 +1533,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,7 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    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 +81,7 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(3 == locator.getAll().size());
+    assertEquals(3, locator.getAll().size());
 
     // when
     conn.setCacheNodesChange("0.0.0.0:11212");
@@ -90,7 +90,7 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    assertEquals(1, locator.getAll().size());
   }
 
   public void testNodesChangeQueue_empty() throws Exception {
@@ -101,7 +101,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 +128,7 @@ public void testNodesChangeQueue_redundent() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(2 == locator.getAll().size());
+    assertEquals(2, locator.getAll().size());
   }
 
   public void testNodesChangeQueue_twice() throws Exception {