Skip to content

Commit a9559bb

Browse files
uhm0311uhm0311
committed
INTERNAL: Not to cancel operations when node is removed from ZK but still alive.
1 parent ba93538 commit a9559bb

11 files changed

Lines changed: 129 additions & 5 deletions

‎src/main/java/net/spy/memcached/ArcusKetamaNodeLocator.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ public class ArcusKetamaNodeLocator extends SpyObject implements NodeLocator {
4242

4343
private final TreeMap<Long, SortedSet<MemcachedNode>> ketamaNodes;
4444
private final Collection<MemcachedNode> allNodes;
45+
private final Collection<MemcachedNode> delayedClosingNodes = new HashSet<MemcachedNode>();
4546

4647
/* ENABLE_MIGRATION if */
4748
private TreeMap<Long, SortedSet<MemcachedNode>> ketamaAlterNodes;
@@ -226,6 +227,10 @@ public void update(Collection<MemcachedNode> toAttach,
226227
for (MemcachedNode node : toDelete) {
227228
allNodes.remove(node);
228229
removeHash(node);
230+
if (node.hasOp() && node.isActive()) {
231+
delayedClosingNodes.add(node);
232+
continue;
233+
}
229234
try {
230235
node.closeChannel();
231236
} catch (IOException e) {
@@ -244,6 +249,14 @@ public void update(Collection<MemcachedNode> toAttach,
244249
}
245250
}
246251

252+
public Collection<MemcachedNode> getDelayedClosingNodes() {
253+
return Collections.unmodifiableCollection(delayedClosingNodes);
254+
}
255+
256+
public void updateDelayedClosingNodes(Collection<MemcachedNode> closedNodes) {
257+
delayedClosingNodes.removeAll(closedNodes);
258+
}
259+
247260
private Long getKetamaHashPoint(byte[] digest, int h) {
248261
return ((long) (digest[3 + h * 4] & 0xFF) << 24)
249262
| ((long) (digest[2 + h * 4] & 0xFF) << 16)
@@ -440,6 +453,10 @@ public void updateAlter(Collection<MemcachedNode> toAttach,
440453
for (MemcachedNode node : toDelete) {
441454
alterNodes.remove(node);
442455
removeHashOfAlter(node);
456+
if (node.hasOp() && node.isActive()) {
457+
delayedClosingNodes.add(node);
458+
continue;
459+
}
443460
try {
444461
node.closeChannel();
445462
} catch (IOException e) {

‎src/main/java/net/spy/memcached/ArcusReplKetamaNodeLocator.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ public class ArcusReplKetamaNodeLocator extends SpyObject implements NodeLocator
4343
private final TreeMap<Long, SortedSet<MemcachedReplicaGroup>> ketamaGroups;
4444
private final HashMap<String, MemcachedReplicaGroup> allGroups;
4545
private final Collection<MemcachedNode> allNodes;
46+
private final Collection<MemcachedNode> delayedClosingNodes = new HashSet<MemcachedNode>();
4647

4748
/* ENABLE_MIGRATION if */
4849
private TreeMap<Long, SortedSet<MemcachedReplicaGroup>> ketamaAlterGroups;
@@ -267,6 +268,10 @@ public void update(Collection<MemcachedNode> toAttach,
267268
for (MemcachedNode node : toDelete) {
268269
allNodes.remove(node);
269270
removeNodeFromGroup(node);
271+
if (node.hasOp() && node.isActive()) {
272+
delayedClosingNodes.add(node);
273+
continue;
274+
}
270275
try {
271276
node.closeChannel();
272277
} catch (IOException e) {
@@ -303,6 +308,14 @@ public void update(Collection<MemcachedNode> toAttach,
303308
}
304309
}
305310

311+
public Collection<MemcachedNode> getDelayedClosingNodes() {
312+
return Collections.unmodifiableCollection(delayedClosingNodes);
313+
}
314+
315+
public void updateDelayedClosingNodes(Collection<MemcachedNode> closedNodes) {
316+
delayedClosingNodes.removeAll(closedNodes);
317+
}
318+
306319
public void switchoverReplGroup(MemcachedReplicaGroup group) {
307320
lock.lock();
308321
group.changeRole();
@@ -573,6 +586,10 @@ public void updateAlter(Collection<MemcachedNode> toAttach,
573586
removeHashOfAlter(mrg);
574587
}
575588
}
589+
if (node.hasOp() && node.isActive()) {
590+
delayedClosingNodes.add(node);
591+
continue;
592+
}
576593
try {
577594
node.closeChannel();
578595
} catch (IOException e) {

‎src/main/java/net/spy/memcached/ArrayModNodeLocator.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import java.util.ArrayList;
2121
import java.util.Arrays;
2222
import java.util.Collection;
23+
import java.util.HashSet;
2324
import java.util.Iterator;
2425
import java.util.List;
2526

@@ -76,6 +77,14 @@ public void update(Collection<MemcachedNode> toAttach, Collection<MemcachedNode>
7677
throw new UnsupportedOperationException("update not supported");
7778
}
7879

80+
public Collection<MemcachedNode> getDelayedClosingNodes() {
81+
return new HashSet<MemcachedNode>();
82+
}
83+
84+
public void updateDelayedClosingNodes(Collection<MemcachedNode> closedNodes) {
85+
// do NOT throw UnsupportedOperationException here for test codes.
86+
}
87+
7988
/* ENABLE_MIGRATION if */
8089
public Collection<MemcachedNode> getAlterAll() {
8190
return new ArrayList<MemcachedNode>();

‎src/main/java/net/spy/memcached/KetamaNodeLocator.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import java.util.ArrayList;
2121
import java.util.Collection;
2222
import java.util.Collections;
23+
import java.util.HashSet;
2324
import java.util.Iterator;
2425
import java.util.List;
2526
import java.util.Map;
@@ -148,6 +149,14 @@ public void update(Collection<MemcachedNode> toAttach, Collection<MemcachedNode>
148149
throw new UnsupportedOperationException("update not supported");
149150
}
150151

152+
public Collection<MemcachedNode> getDelayedClosingNodes() {
153+
return new HashSet<MemcachedNode>();
154+
}
155+
156+
public void updateDelayedClosingNodes(Collection<MemcachedNode> closedNodes) {
157+
// do NOT throw UnsupportedOperationException here for test codes.
158+
}
159+
151160
public SortedMap<Long, MemcachedNode> getKetamaNodes() {
152161
return Collections.unmodifiableSortedMap(ketamaNodes);
153162
}

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

Lines changed: 47 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -304,6 +304,9 @@ public void handleIO() throws IOException {
304304
}
305305
/* ENABLE_REPLICATION end */
306306

307+
// Deal with the memcached nodes that removed from ZK but has operation in queue.
308+
handleDelayedClosingNodes();
309+
307310
// Deal with the memcached server group that's been added by CacheManager.
308311
handleCacheNodesChange();
309312

@@ -330,12 +333,14 @@ private void handleNodesToRemove(final List<MemcachedNode> nodesToRemove) {
330333
}
331334
/* ENABLE_MIGRATION end */
332335

336+
if (node.isActive()) {
337+
// if memcached node is removed from ZK but still can serve operations, do NOT cancel it.
338+
continue;
339+
}
340+
333341
// removing node is not related to failure mode.
334342
// so, cancel operations regardless of failure mode.
335-
String cause = "node removed.";
336-
cancelOperations(node.destroyReadQueue(false), cause);
337-
cancelOperations(node.destroyWriteQueue(false), cause);
338-
cancelOperations(node.destroyInputQueue(), cause);
343+
cancelAllOperations(node, "node removed.");
339344
}
340345
}
341346

@@ -680,6 +685,38 @@ public void complete() {
680685
addOperation(node, op);
681686
}
682687

688+
// Handle the memcached nodes that removed from ZK but has operation in queue.
689+
void handleDelayedClosingNodes() {
690+
Collection<MemcachedNode> closingNodes = locator.getDelayedClosingNodes();
691+
if (closingNodes.isEmpty()) {
692+
return;
693+
}
694+
695+
Collection<MemcachedNode> closedNodes = new HashSet<MemcachedNode>();
696+
for (MemcachedNode node : closingNodes) {
697+
boolean isActive = node.isActive();
698+
boolean hasOp = node.hasOp();
699+
700+
if (isActive && !hasOp) {
701+
try {
702+
node.closeChannel();
703+
} catch (IOException e) {
704+
getLogger().error("Failed to closeChannel the node : " + node);
705+
}
706+
} else if (!isActive && hasOp) {
707+
cancelAllOperations(node, "connection lost after node removed.");
708+
} else {
709+
continue;
710+
}
711+
712+
closedNodes.add(node);
713+
}
714+
715+
if (!closedNodes.isEmpty()) {
716+
locator.updateDelayedClosingNodes(closedNodes);
717+
}
718+
}
719+
683720
// Handle the memcached server group that's been added by CacheManager.
684721
void handleCacheNodesChange() throws IOException {
685722
/* ENABLE_MIGRATION if */
@@ -1225,6 +1262,12 @@ private void cancelOperations(Collection<Operation> ops, String cause) {
12251262
}
12261263
}
12271264

1265+
private void cancelAllOperations(MemcachedNode node, String cause) {
1266+
cancelOperations(node.destroyReadQueue(false), cause);
1267+
cancelOperations(node.destroyWriteQueue(false), cause);
1268+
cancelOperations(node.destroyInputQueue(), cause);
1269+
}
1270+
12281271
private void redistributeOperations(Collection<Operation> ops, String cause) {
12291272
for (Operation op : ops) {
12301273
if (op instanceof KeyedOperation) {

‎src/main/java/net/spy/memcached/MemcachedNode.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,11 @@ public interface MemcachedNode {
109109
*/
110110
boolean hasWriteOp();
111111

112+
/**
113+
* True if any operation is in operation queue.
114+
*/
115+
boolean hasOp();
116+
112117
/**
113118
* Add an operation to the queue. Authentication operations should
114119
* never be added to the queue, but this is not checked.

‎src/main/java/net/spy/memcached/MemcachedNodeROImpl.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,10 @@ public boolean hasWriteOp() {
125125
return root.hasReadOp();
126126
}
127127

128+
public boolean hasOp() {
129+
return root.hasOp();
130+
}
131+
128132
public boolean isActive() {
129133
return root.isActive();
130134
}

‎src/main/java/net/spy/memcached/NodeLocator.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,18 @@ public interface NodeLocator {
5858
*/
5959
void update(Collection<MemcachedNode> toAttach, Collection<MemcachedNode> toDelete);
6060

61+
/**
62+
* Get all memcached nodes that removed from ZK but has operation in queue.
63+
* Note that this feature is only available in ArcusKetamaNodeLocator.
64+
*/
65+
Collection<MemcachedNode> getDelayedClosingNodes();
66+
67+
/**
68+
* Update all memcached nodes that removed from ZK but has operation in queue.
69+
* Note that this feature is only available in ArcusKetamaNodeLocator.
70+
*/
71+
void updateDelayedClosingNodes(Collection<MemcachedNode> closedNodes);
72+
6173
/* ENABLE_MIGRATION if */
6274
/**
6375
* Get all alter memcached nodes.

‎src/main/java/net/spy/memcached/protocol/TCPMemcachedNodeImpl.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -343,6 +343,10 @@ public final boolean hasWriteOp() {
343343
return !(optimizedOp == null && writeQ.isEmpty());
344344
}
345345

346+
public final boolean hasOp() {
347+
return hasReadOp() || hasWriteOp() || !inputQueue.isEmpty();
348+
}
349+
346350
public final void addOpToInputQ(Operation op) {
347351
op.setHandlingNode(this);
348352
op.initialize();

‎src/test/java/net/spy/memcached/MemcachedNodeROImplTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ public void testReadOnliness() throws Exception {
2929
Set<String> acceptable = new HashSet<String>(Arrays.asList(
3030
"toString", "getSocketAddress", "getBytesRemainingToWrite",
3131
"getReconnectCount", "getSelectionOps", "getNodeName", "hasReadOp",
32-
"hasWriteOp", "isActive", "isFirstConnecting"));
32+
"hasWriteOp", "hasOp", "isActive", "isFirstConnecting"));
3333

3434
for (Method meth : MemcachedNode.class.getMethods()) {
3535
if (acceptable.contains(meth.getName())) {

0 commit comments

Comments
 (0)