Skip to content

Make sure ConsistentHashingNodeProvider returns unique candidates #17228

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rongrong merged 1 commit into from
Feb 7, 2022
Merged

Make sure ConsistentHashingNodeProvider returns unique candidates #17228

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
46 changes: 36 additions & 10 deletions .../src/main/java/com/facebook/presto/execution/scheduler/ConsistentHashingNodeProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,12 @@
import com.google.common.hash.HashFunction;

import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;

import static com.facebook.presto.common.type.encoding.StringUtils.UTF_8;
Expand All @@ -35,6 +38,7 @@ public class ConsistentHashingNodeProvider
{
private static final HashFunction HASH_FUNCTION = murmur3_32();
private final NavigableMap<Integer, InternalNode> candidates;
private final int nodeCount;

static ConsistentHashingNodeProvider create(Collection<InternalNode> nodes, int weight)
{
Expand All @@ -44,27 +48,49 @@ static ConsistentHashingNodeProvider create(Collection<InternalNode> nodes, int
activeNodesByConsistentHashing.put(murmur3_32().hashString(format("%s%d", node.getNodeIdentifier(), i), UTF_8).asInt(), node);
}
}
return new ConsistentHashingNodeProvider(activeNodesByConsistentHashing);
return new ConsistentHashingNodeProvider(activeNodesByConsistentHashing, nodes.size());
}

public ConsistentHashingNodeProvider(NavigableMap<Integer, InternalNode> candidates)
private ConsistentHashingNodeProvider(NavigableMap<Integer, InternalNode> candidates, int nodeCount)
{
this.candidates = requireNonNull(candidates, "candidates is null");
this.nodeCount = nodeCount;
}

@Override
public List<HostAddress> get(String key, int count)
{
if (count > nodeCount) {
count = nodeCount;
}
ImmutableList.Builder<HostAddress> nodes = ImmutableList.builder();
for (int i = 0; i < count; i++) {
int hashKey = HASH_FUNCTION.hashString(format("%s%d", key, i), UTF_8).asInt();
Map.Entry<Integer, InternalNode> entry = candidates.ceilingEntry(hashKey);
if (entry != null) {
nodes.add(candidates.ceilingEntry(hashKey).getValue().getHostAndPort());
}
else {
nodes.add(candidates.firstEntry().getValue().getHostAndPort());
Set<HostAddress> unique = new HashSet<>();
int hashKey = HASH_FUNCTION.hashString(format("%s", key), UTF_8).asInt();
Map.Entry<Integer, InternalNode> entry = candidates.ceilingEntry(hashKey);
HostAddress candidate;
SortedMap<Integer, InternalNode> nextEntries;
if (entry != null) {
candidate = entry.getValue().getHostAndPort();
nextEntries = candidates.tailMap(entry.getKey(), false);
}
else {
candidate = candidates.firstEntry().getValue().getHostAndPort();
nextEntries = candidates.tailMap(candidates.firstKey(), false);
}
unique.add(candidate);
nodes.add(candidate);
while (unique.size() < count) {
Copy link
Copy Markdown
Contributor

@yuanzhanhku yuanzhanhku Jan 29, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could this be an infinite loop? e.g., when num of workers is < count? Add a test for this case?

It is also possible that this will take a lot of iterations to find next unique candidate if the hash function happens to return a visited virtual node for a lot of times.
How about searching it in the candidates map to find the next unique candidate instead of relying on a random function? Specifically, if we find a duplicate node, we get its iterator/pointer in the map and check the following nodes according to their order in the map until we find a new node or have exhausted all node.

Copy link
Copy Markdown
Contributor Author

@rongrong rongrong Jan 30, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, I think the assumption here is that count is small. 2 is used right now. But I agree we should add a check. I think in any realistic scenario the probability of continuously hitting the same virtual nodes is pretty low but that's also assuming count is small. Next physical node on the ring seems better.

for (Map.Entry<Integer, InternalNode> next : nextEntries.entrySet()) {
candidate = next.getValue().getHostAndPort();
if (!unique.contains(candidate)) {
unique.add(candidate);
nodes.add(candidate);
if (unique.size() == count) {
break;
}
}
}
nextEntries = candidates;
}
return nodes.build();
}
Expand Down
3 changes: 3 additions & 0 deletions ...ain/src/main/java/com/facebook/presto/execution/scheduler/ModularHashingNodeProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ public List<HostAddress> get(String identifier, int count)
int mod = identifier.hashCode() % size;
int position = mod < 0 ? mod + size : mod;
List<HostAddress> chosenCandidates = new ArrayList<>();
if (count > size) {
count = size;
}
for (int i = 0; i < count && i < sortedCandidates.size(); i++) {
chosenCandidates.add(sortedCandidates.get((position + i) % size).getHostAndPort());
}
Expand Down
63 changes: 32 additions & 31 deletions presto-main/src/test/java/com/facebook/presto/execution/TestNodeScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@
import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.HARD_AFFINITY;
import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.NO_PREFERENCE;
import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
import static java.util.concurrent.Executors.newCachedThreadPool;
import static java.util.concurrent.Executors.newScheduledThreadPool;
Expand Down Expand Up @@ -611,11 +612,11 @@ public void testAffinityAssignmentWithConsistentHashing()
// In setup node 1-3 are added to node manager
SplitPlacementResult splitPlacementResult = nodeSelector.computeAssignments(splits, ImmutableList.of());
assertEquals(splitPlacementResult.getAssignments().keySet().size(), 3);
// node1: split 2, 3, 5, 6, 9
// node1: split 1, 3, 4, 5, 6, 7, 8, 9
Collection<ConnectorSplit> node1Splits = splitPlacementResult.getAssignments().get(node1).stream().map(Split::getConnectorSplit).collect(toImmutableSet());
// node2: split 1, 7
// node2: 0
Collection<Object> node2Splits = splitPlacementResult.getAssignments().get(node2).stream().map(Split::getConnectorSplit).collect(toImmutableSet());
// node3: split 0, 4, 8
// node3: split 2
Collection<Object> node3Splits = splitPlacementResult.getAssignments().get(node3).stream().map(Split::getConnectorSplit).collect(toImmutableSet());

// Scheduling the same splits on the same set of nodes should give the same assignment
Expand Down Expand Up @@ -669,22 +670,22 @@ public void testAffinityAssignmentWithConsistentHashingWithVirtualNodes()
// entry5 ( 2145381619): node1
SplitPlacementResult splitPlacementResult = nodeSelector.computeAssignments(splits, ImmutableList.of());
// hashing value for splits:
// 0: -1879591379 -> entry1 -> node3
// 1: -2031875777 -> entry0 -> node2
// 2: -163077544 -> entry3 -> node3
// 3: 749129358 -> entry3 -> node3
// 4: -1784631546 -> entry1 -> node3
// 5: -118156056 -> entry3 -> node3
// 6: 388471277 -> entry3 -> node3
// 7: -2084245305 -> entry0 -> node2
// 8: -1127017311 -> entry1 -> node3
// 9: 1305218356 -> entry5 -> node1
// 0: -1962219106 -> entry0 -> node2
// 1: 145569539 -> entry3 -> node3
// 2: -1599101205 -> entry1 -> node3
// 3: -165119218 -> entry3 -> node3
// 4: 1142216720 -> entry4 -> node1
// 5: 1347620135 -> entry5 -> node1
// 6: 1232195252 -> entry5 -> node1
// 7: 427886318 -> entry3 -> node3
// 8: 1469878697 -> entry5 -> node1
// 9: 296801082 -> entry3 -> node3
assertEquals(splitPlacementResult.getAssignments().keySet().size(), 3);
// node1: split 9
// node1: split 4, 5, 6, 8
Collection<ConnectorSplit> node1Splits = splitPlacementResult.getAssignments().get(node1).stream().map(Split::getConnectorSplit).collect(toImmutableSet());
// node2: split 1, 7
// node2: split 0
Collection<Object> node2Splits = splitPlacementResult.getAssignments().get(node2).stream().map(Split::getConnectorSplit).collect(toImmutableSet());
// node3: split 0, 2, 3, 4, 5, 6, 8
// node3: split 1, 2, 3, 7, 9
Collection<Object> node3Splits = splitPlacementResult.getAssignments().get(node3).stream().map(Split::getConnectorSplit).collect(toImmutableSet());

// Scheduling the same splits on the same set of nodes should give the same assignment
Expand All @@ -707,21 +708,21 @@ public void testAffinityAssignmentWithConsistentHashingWithVirtualNodes()
nodeSelector = nodeScheduler.createNodeSelector(session, CONNECTOR_ID, 3);
splitPlacementResult = nodeSelector.computeAssignments(splits, ImmutableList.of());
// hashing value for splits:
// 0: -1879591379 -> entry1 -> node4
// 1: -2031875777 -> entry0 -> node2
// 2: -163077544 -> entry4 -> node3
// 3: 749129358 -> entry4 -> node3
// 4: -1784631546 -> entry1 -> node4
// 5: -118156056 -> entry4 -> node3
// 6: 388471277 -> entry4 -> node3
// 7: -2084245305 -> entry0 -> node2
// 8: -1127017311 -> entry2 -> node3
// 9: 1305218356 -> entry6 -> node4
assertEquals(splitPlacementResult.getAssignments().keySet().size(), 3);
assertEquals(splitPlacementResult.getAssignments().get(node1), ImmutableSet.of());
// 0: -1962219106 -> entry0 -> node2
// 1: 145569539 -> entry4 -> node3
// 2: -1599101205 -> entry2 -> node3
// 3: -165119218 -> entry4 -> node3
// 4: 1142216720 -> entry5 -> node1
// 5: 1347620135 -> entry6 -> node4
// 6: 1232195252 -> entry6 -> node4
// 7: 427886318 -> entry4 -> node3
// 8: 1469878697 -> entry6 -> node4
// 9: 296801082 -> entry4 -> node3
assertEquals(splitPlacementResult.getAssignments().keySet().size(), 4);
assertEquals(splitPlacementResult.getAssignments().get(node1).stream().map(Split::getConnectorSplit).map(ConnectorSplit::getSplitIdentifier).collect(toImmutableSet()), ImmutableSet.of(4));
assertEquals(splitPlacementResult.getAssignments().get(node2).stream().map(Split::getConnectorSplit).collect(toImmutableSet()), node2Splits);
assertEquals(splitPlacementResult.getAssignments().get(node3).stream().map(Split::getConnectorSplit).map(ConnectorSplit::getSplitIdentifier).collect(toImmutableSet()), ImmutableSet.of(2, 3, 5, 6, 8));
assertEquals(splitPlacementResult.getAssignments().get(node4).stream().map(Split::getConnectorSplit).map(ConnectorSplit::getSplitIdentifier).collect(toImmutableSet()), ImmutableSet.of(0, 4, 9));
assertEquals(splitPlacementResult.getAssignments().get(node3).stream().map(Split::getConnectorSplit).map(ConnectorSplit::getSplitIdentifier).collect(toImmutableSet()), ImmutableSet.of(1, 2, 3, 7, 9));
assertEquals(splitPlacementResult.getAssignments().get(node4).stream().map(Split::getConnectorSplit).map(ConnectorSplit::getSplitIdentifier).collect(toImmutableSet()), ImmutableSet.of(5, 6, 8));
}

@Test
Expand Down Expand Up @@ -1315,7 +1316,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
@Override
public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
{
return nodeProvider.get(String.valueOf(scheduleIdentifierId), 1);
return nodeProvider.get(format("split%d", scheduleIdentifierId), 1);
}

@Override
Expand Down
20 changes: 19 additions & 1 deletion .../test/java/com/facebook/presto/execution/scheduler/TestConsistentHashingNodeProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import com.facebook.presto.client.NodeVersion;
import com.facebook.presto.metadata.InternalNode;
import com.facebook.presto.spi.HostAddress;
import com.google.common.collect.ImmutableSet;
import org.testng.annotations.Test;

import java.net.URI;
Expand All @@ -26,7 +27,10 @@
import java.util.stream.IntStream;

import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static java.lang.String.format;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertTrue;

public class TestConsistentHashingNodeProvider
Expand All @@ -39,10 +43,24 @@ public void testDistribution()
Random random = new Random();
Map<HostAddress, Integer> result = new HashMap<>();
for (int i = 0; i < 1_000_000; i++) {
HostAddress hostAddress = nodeProvider.get(format("split%d", random.nextInt()), 1).get(0);
List<HostAddress> candidates = nodeProvider.get(format("split%d", random.nextInt()), 2);
assertNotEquals(candidates.get(1), candidates.get(0));
HostAddress hostAddress = candidates.get(0);
int count = result.getOrDefault(hostAddress, 0);
result.put(hostAddress, count + 1);
}
assertTrue(result.values().stream().allMatch(count -> count >= 80000 && count <= 120000));
}

@Test
public void testMultipleCandidates()
{
List<InternalNode> nodes = IntStream.range(0, 10).mapToObj(i -> new InternalNode(format("other%d", i), URI.create(format("http://127.0.0.%d:100", i)), NodeVersion.UNKNOWN, false)).collect(toImmutableList());
ConsistentHashingNodeProvider nodeProvider = ConsistentHashingNodeProvider.create(nodes, 1);
assertEquals(ImmutableSet.copyOf(nodeProvider.get("split1", 10)), nodes.stream().map(InternalNode::getHostAndPort).collect(toImmutableSet()));
assertEquals(ImmutableSet.copyOf(nodeProvider.get("split1", 11)), nodes.stream().map(InternalNode::getHostAndPort).collect(toImmutableSet()));

ConsistentHashingNodeProvider nodeProviderWithWeight = ConsistentHashingNodeProvider.create(nodes, 100);
assertEquals(ImmutableSet.copyOf(nodeProvider.get("split1", 10)), nodes.stream().map(InternalNode::getHostAndPort).collect(toImmutableSet()));
}
}