Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions src/main/java/net/spy/memcached/ArcusReplNodeAddress.java
Original file line number Diff line number Diff line change
Expand Up @@ -109,8 +109,7 @@ static List<InetSocketAddress> getAddresses(String s) {
return list;
}

static Map<String, List<ArcusReplNodeAddress>> makeGroupAddrsList(
List<InetSocketAddress> addrs) {
static Map<String, List<ArcusReplNodeAddress>> makeGroupAddrs(List<InetSocketAddress> addrs) {

Map<String, List<ArcusReplNodeAddress>> newAllGroups =
new HashMap<>();
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/net/spy/memcached/CacheManager.java
Original file line number Diff line number Diff line change
Expand Up @@ -656,7 +656,7 @@ private String getInfo() {
/* ENABLE_REPLICATION if */
private List<InetSocketAddress> validateReplicaGroup(List<InetSocketAddress> socketList) {
Map<String, List<ArcusReplNodeAddress>> newAllGroups =
ArcusReplNodeAddress.makeGroupAddrsList(socketList);
ArcusReplNodeAddress.makeGroupAddrs(socketList);

// recreate socket list
socketList.clear();
Expand Down
99 changes: 28 additions & 71 deletions src/main/java/net/spy/memcached/MemcachedConnection.java
Original file line number Diff line number Diff line change
Expand Up @@ -305,7 +305,7 @@ public void handleIO() throws IOException {
/* ENABLE_REPLICATION if */
if (arcusReplEnabled) {
// Deal with the memcached server group that need delayed switchover.
handleDelayedSwitchover();
handleDelayedSwitchover(false);
}
/* ENABLE_REPLICATION end */

Expand Down Expand Up @@ -369,95 +369,52 @@ private void updateConnections(List<InetSocketAddress> addrs) throws IOException
}

/* ENABLE_REPLICATION if */
private Set<String> findChangedGroups(List<InetSocketAddress> addrs,
Collection<MemcachedNode> nodes) {
Map<String, InetSocketAddress> addrMap = new HashMap<>();
for (InetSocketAddress each : addrs) {
addrMap.put(each.toString(), each);
}

Set<String> changedGroupSet = new HashSet<>();
for (MemcachedNode node : nodes) {
String nodeAddr = ((InetSocketAddress) node.getSocketAddress()).toString();
if (addrMap.remove(nodeAddr) == null) { // removed node
changedGroupSet.add(node.getReplicaGroup().getGroupName());
}
}
for (String addr : addrMap.keySet()) { // newly added node
ArcusReplNodeAddress a = (ArcusReplNodeAddress) addrMap.get(addr);
changedGroupSet.add(a.getGroupName());
}
return changedGroupSet;
}

private List<InetSocketAddress> findAddrsOfChangedGroups(List<InetSocketAddress> addrs,
Set<String> changedGroups) {
List<InetSocketAddress> changedGroupAddrs = new ArrayList<>();
for (InetSocketAddress addr : addrs) {
if (changedGroups.contains(((ArcusReplNodeAddress) addr).getGroupName())) {
changedGroupAddrs.add(addr);
}
}
return changedGroupAddrs;
}

private void updateReplConnections(List<InetSocketAddress> addrs) throws IOException {
List<MemcachedNode> attachNodes = new ArrayList<>();
List<MemcachedNode> removeNodes = new ArrayList<>();
List<MemcachedReplicaGroup> changeRoleGroups = new ArrayList<>();
List<Task> taskList = new ArrayList<>(); // tasks executed after locator update

/* In replication, after SWITCHOVER or REPL_SLAVE is received from a group
* and switchover is performed, but before the group's znode is changed,
* another group's znode can be changed.
*
* In this case, there is a problem that the switchover is restored
* because the state of the switchover group and the znode state are different.
*
* In order to remove the abnormal phenomenon,
* we find out the changed groups with the comparison of previous and current znode list,
* and update the state of groups based on them.
*/
Set<String> changedGroups = findChangedGroups(addrs, locator.getAll());
// Create new group list from the provided addresses
Map<String, List<ArcusReplNodeAddress>> newGroups = ArcusReplNodeAddress.makeGroupAddrs(addrs);
// Get the existing groups from the locator
Map<String, MemcachedReplicaGroup> oldGroups =
((ArcusReplKetamaNodeLocator) locator).getAllGroups();
Comment thread
brido4125 marked this conversation as resolved.
Set<String> invalidGroups = new HashSet<>();

Map<String, List<ArcusReplNodeAddress>> newAllGroups =
ArcusReplNodeAddress.makeGroupAddrsList(findAddrsOfChangedGroups(addrs, changedGroups));
// Immediately exec the previous delayed switchover case.
handleDelayedSwitchover(true);
Comment thread
oliviarla marked this conversation as resolved.

// remove invalidated groups in changedGroups
for (Map.Entry<String, List<ArcusReplNodeAddress>> entry : newAllGroups.entrySet()) {
for (Map.Entry<String, List<ArcusReplNodeAddress>> entry : newGroups.entrySet()) {
if (!ArcusReplNodeAddress.validateGroup(entry)) {
changedGroups.remove(entry.getKey());
invalidGroups.add(entry.getKey());
continue;
}
// Handle newly added groups
if (!oldGroups.containsKey(entry.getKey())) {
for (ArcusReplNodeAddress newAddr : entry.getValue()) {
attachNodes.add(attachMemcachedNode(newAddr));
}
}
}

Map<String, MemcachedReplicaGroup> oldAllGroups =
((ArcusReplKetamaNodeLocator) locator).getAllGroups();

for (String changedGroupName : changedGroups) {
MemcachedReplicaGroup oldGroup = oldAllGroups.get(changedGroupName);
List<ArcusReplNodeAddress> newGroupAddrs = newAllGroups.get(changedGroupName);
for (Map.Entry<String, MemcachedReplicaGroup> oldGroupEntry : oldGroups.entrySet()) {
String groupName = oldGroupEntry.getKey();
MemcachedReplicaGroup oldGroup = oldGroupEntry.getValue();
List<ArcusReplNodeAddress> newGroupAddrs = newGroups.get(groupName);

if (oldGroup == null) {
// Newly added group
for (ArcusReplNodeAddress newAddr : newGroupAddrs) {
attachNodes.add(attachMemcachedNode(newAddr));
}
// If group name exists in old groups, invalid case is ignored.
if (invalidGroups.contains(groupName)) {
Comment thread
jhpark816 marked this conversation as resolved.
continue;
}

if (newGroupAddrs == null) {
// Old group nodes have disappeared. Remove the old group nodes.
removeNodes.add(oldGroup.getMasterNode());
removeNodes.addAll(oldGroup.getSlaveNodes());
delayedSwitchoverGroups.remove(oldGroup);
continue;
}

if (oldGroup.isDelayedSwitchover()) {
delayedSwitchoverGroups.remove(oldGroup);
switchoverMemcachedReplGroup(oldGroup, true);
}

MemcachedNode oldMasterNode = oldGroup.getMasterNode();
List<MemcachedNode> oldSlaveNodes = oldGroup.getSlaveNodes();

Expand Down Expand Up @@ -846,9 +803,9 @@ private void updateAlterConnections(List<InetSocketAddress> addrs) throws IOExce
/* ENABLE_MIGRATION end */

// Handle the memcached server group that need delayed switchover.
private void handleDelayedSwitchover() {
private void handleDelayedSwitchover(boolean immediate) {
if (!delayedSwitchoverGroups.isEmpty()) {
delayedSwitchoverGroups.switchover();
delayedSwitchoverGroups.switchover(immediate);
}
}

Expand Down Expand Up @@ -1793,15 +1750,15 @@ public long getMinDelayMillis() {
1);
}

public void switchover() {
public void switchover(boolean immediate) {
long now = System.nanoTime();
Iterator<Entry<Long, MemcachedReplicaGroup>> iterator = groups.entrySet().iterator();
while (iterator.hasNext()) {
Entry<Long, MemcachedReplicaGroup> entry = iterator.next();
long switchoverTime = entry.getKey();
MemcachedReplicaGroup group = entry.getValue();

if (now < switchoverTime) {
if (!immediate && now < switchoverTime) {
return;
} else {
iterator.remove();
Expand Down