ENHANCE: Create hashUpdateThreadPool to decrease IO thread overhead.

naver · Mar 11, 2024 · b7112f0 · b7112f0
1 parent e912341
commit b7112f0
Show file tree

Hide file tree

Showing 4 changed files with 66 additions and 12 deletions.
diff --git a/src/main/java/net/spy/memcached/CacheListUpdateTask.java b/src/main/java/net/spy/memcached/CacheListUpdateTask.java
@@ -0,0 +1,27 @@
+package net.spy.memcached;
+
+import java.util.List;
+
+public class CacheListUpdateTask implements Runnable {
+
+  private final MemcachedConnection conn;
+  private final List<MemcachedNode> attachNodes;
+  private final List<MemcachedNode> removeNodes;
+
+  public CacheListUpdateTask(MemcachedConnection memcachedConnection,
+                             List<MemcachedNode> attachNodes,
+                             List<MemcachedNode> removeNodes) {
+
+    this.conn = memcachedConnection;
+    this.attachNodes = attachNodes;
+    this.removeNodes = removeNodes;
+  }
+
+  @Override
+  public void run() {
+    conn.getLocator().update(attachNodes, removeNodes);
+
+    // Remove the unavailable nodes.
+    conn.handleNodesToRemove(removeNodes);
+  }
+}
diff --git a/src/main/java/net/spy/memcached/HashRingUpdateService.java b/src/main/java/net/spy/memcached/HashRingUpdateService.java
@@ -0,0 +1,20 @@
+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 void updateHashes(CacheListUpdateTask task) {
+    pool.execute(task);
+  }
+
+  public void shutdown() {
+    pool.shutdown();
+  }
+}
diff --git a/src/main/java/net/spy/memcached/MemcachedConnection.java b/src/main/java/net/spy/memcached/MemcachedConnection.java
@@ -114,6 +114,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.
@@ -312,7 +313,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);
@@ -357,10 +358,7 @@ 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 */
@@ -1509,6 +1507,7 @@ public void shutdown() throws IOException {
       }
     }
     selector.close();
+    hashUpdateService.shutdown();
     getLogger().debug("Shut down selector %s", selector);
   }
 

diff --git a/src/test/java/net/spy/memcached/MemcachedConnectionTest.java b/src/test/java/net/spy/memcached/MemcachedConnectionTest.java
@@ -64,6 +64,7 @@ public void testDebugBuffer() {
     assertEquals("this is a test \\x5f", s);
   }
 
+  // FIXME : Change to using callback instead of sleep when CompletableFuture enabled.
   public void testNodesChangeQueue() throws Exception {
     // when
     conn.setCacheNodesChange("0.0.0.0:11211");
@@ -72,7 +73,8 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    Thread.sleep(500);
+    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 +83,8 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(3 == locator.getAll().size());
+    Thread.sleep(500);
+    assertEquals(3, locator.getAll().size());
 
     // when
     conn.setCacheNodesChange("0.0.0.0:11212");
@@ -90,7 +93,8 @@ public void testNodesChangeQueue() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    Thread.sleep(500);
+    assertEquals(1, locator.getAll().size());
   }
 
   public void testNodesChangeQueue_empty() throws Exception {
@@ -101,7 +105,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() {
@@ -120,17 +124,20 @@ public void testNodesChangeQueue_invalid_addr() {
     }
   }
 
-  public void testNodesChangeQueue_redundant() throws Exception {
+  // FIXME : Change to using callback instead of sleep when CompletableFuture enabled.
+  public void testNodesChangeQueue_redundent() throws Exception {
     // when
     conn.setCacheNodesChange("0.0.0.0:11211,0.0.0.0:11211");
 
     // test
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(2 == locator.getAll().size());
+    Thread.sleep(500);
+    assertEquals(2, locator.getAll().size());
   }
 
+  // FIXME : Change to using callback instead of sleep when CompletableFuture enabled.
   public void testNodesChangeQueue_twice() throws Exception {
     // when
     conn.setCacheNodesChange("0.0.0.0:11211");
@@ -140,7 +147,8 @@ public void testNodesChangeQueue_twice() throws Exception {
     conn.handleCacheNodesChange();
 
     // then
-    assertTrue(1 == locator.getAll().size());
+    Thread.sleep(500);
+    assertEquals(1, locator.getAll().size());
   }
 
   public void testAddOperations() throws Exception {