Skip to content

Commit

Permalink
ENHANCE: Create hashUpdateThreadPool to decrease IO thread overhead.
Browse files Browse the repository at this point in the history
  • Loading branch information
brido4125 committed Aug 7, 2023
1 parent c1b9b96 commit 9a955aa
Show file tree
Hide file tree
Showing 4 changed files with 67 additions and 10 deletions.
30 changes: 30 additions & 0 deletions src/main/java/net/spy/memcached/CacheListUpdateTask.java
Original file line number Diff line number Diff line change
@@ -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;
}
}
26 changes: 26 additions & 0 deletions src/main/java/net/spy/memcached/HashRingUpdateService.java
Original file line number Diff line number Diff line change
@@ -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();
}
}
11 changes: 6 additions & 5 deletions src/main/java/net/spy/memcached/MemcachedConnection.java
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 */
Expand Down Expand Up @@ -1533,6 +1533,7 @@ public void shutdown() throws IOException {
}
}
selector.close();
hashUpdateService.shutdown();
getLogger().debug("Shut down selector %s", selector);
}

Expand Down
10 changes: 5 additions & 5 deletions src/test/java/net/spy/memcached/MemcachedConnectionTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -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");
Expand All @@ -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 {
Expand All @@ -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() {
Expand All @@ -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 {
Expand Down

0 comments on commit 9a955aa

Please sign in to comment.