Skip to content

HDDS-17553. Avoid NetworkTopologyImpl lock in OM read-path datanode s… #11838

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

Open
greenwich wants to merge 1 commit into
base: master
Choose a base branch
from
Open

HDDS-17553. Avoid NetworkTopologyImpl lock in OM read-path datanode s… #11838

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
72 changes: 47 additions & 25 deletions hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/net/NetworkTopologyImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -715,6 +715,21 @@ private Node chooseNodeInternal(String scope, int leafIndex,
*/
@Override
public int getDistanceCost(Node node1, Node node2) {
netlock.readLock().lock();
try {
return getDistanceCostWithoutLock(node1, node2);
} finally {
netlock.readLock().unlock();
}
}

/**
* Same as {@link #getDistanceCost(Node, Node)} without taking {@link #netlock}, like HDFS
* {@code NetworkTopology#getWeight}. Tree nodes are compared by path, like HDFS {@code NodeBase#equals},
* so no child maps are read. A concurrent topology change can only make the result stale or
* {@link Integer#MAX_VALUE}, which is acceptable for ordering replicas.
*/
private int getDistanceCostWithoutLock(Node node1, Node node2) {
if (Objects.equals(node1, node2)) {
return 0;
}
Expand All @@ -734,35 +749,42 @@ public int getDistanceCost(Node node1, Node node2) {
}

int cost = 0;
netlock.readLock().lock();
try {
Node ancestor1 = node1.getAncestor(level1 - 1);
Node ancestor2 = node2.getAncestor(level2 - 1);
if (!Objects.equals(ancestor1, clusterTree) ||
!Objects.equals(ancestor2, clusterTree)) {
LOG.debug("One of the nodes is outside of network topology");
Node ancestor1 = node1.getAncestor(level1 - 1);
Node ancestor2 = node2.getAncestor(level2 - 1);
if (!isSamePath(ancestor1, clusterTree) || !isSamePath(ancestor2, clusterTree)) {
LOG.debug("One of the nodes is outside of network topology");
return Integer.MAX_VALUE;
}
// A null parent means the node was removed after the ancestor check.
while (level1 > level2) {
node1 = node1.getParent();
if (node1 == null) {
return Integer.MAX_VALUE;
}
while (level1 > level2 && node1 != null) {
node1 = node1.getParent();
level1--;
cost += node1 == null ? 0 : node1.getCost();
}
while (level2 > level1 && node2 != null) {
node2 = node2.getParent();
level2--;
cost += node2 == null ? 0 : node2.getCost();
level1--;
cost += node1.getCost();
}
while (level2 > level1) {
node2 = node2.getParent();
if (node2 == null) {
return Integer.MAX_VALUE;
}
while (node1 != null && node2 != null && !Objects.equals(node1, node2)) {
node1 = node1.getParent();
node2 = node2.getParent();
cost += node1 == null ? 0 : node1.getCost();
cost += node2 == null ? 0 : node2.getCost();
level2--;
cost += node2.getCost();
}
while (!isSamePath(node1, node2)) {
node1 = node1.getParent();
node2 = node2.getParent();
if (node1 == null || node2 == null) {
return Integer.MAX_VALUE;
}
return cost;
} finally {
netlock.readLock().unlock();
cost += node1.getCost() + node2.getCost();
}
return cost;
}

private static boolean isSamePath(Node node1, Node node2) {
return node1 != null && node1.getNetworkFullPath().equals(node2.getNetworkFullPath());
}

/**
Expand Down Expand Up @@ -796,7 +818,7 @@ public <N extends Node> List<N> sortByDistanceCost(Node reader,
// Sort weights for the nodes array
int[] costs = new int[activeLen];
for (int i = 0; i < activeLen; i++) {
costs[i] = getDistanceCost(reader, nodes.get(i));
costs[i] = getDistanceCostWithoutLock(reader, nodes.get(i));
}
// Add cost/node pairs to a TreeMap to sort
NavigableMap<Integer, List<N>> tree = new TreeMap<>();
Expand Down
78 changes: 78 additions & 0 deletions ...-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/net/TestNetworkTopologyImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.util.ArrayList;
import java.util.Arrays;
Expand All @@ -53,6 +54,11 @@
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
Expand Down Expand Up @@ -938,6 +944,78 @@ void testSortByDistanceCostNullReader(NodeSchema[] schemas,
}
}

@ParameterizedTest
@MethodSource("topologies")
void testSortByDistanceCostDoesNotTakeLockPerNode(NodeSchema[] schemas, Node[] nodeArray) {
initNetworkTopology(schemas, nodeArray);
NetworkTopology spyCluster = spy(cluster);
List<Node> nodeList = Arrays.asList(dataNodes);

List<? extends Node> ret = spyCluster.sortByDistanceCost(dataNodes[0], nodeList, nodeList.size());

assertEquals(dataNodes[0], ret.get(0));
verify(spyCluster, never()).getDistanceCost(any(), any());
}

@Test
void testDistanceCostIsMaxWhenNodeIsRemovedDuringWalk() {
initNetworkTopology(new NodeSchema[] {ROOT_SCHEMA, RACK_SCHEMA, LEAF_SCHEMA},
new Node[] {createDatanode("1.1.1.1", "/r1"), createDatanode("2.2.2.2", "/r1")});
// Passes the ancestor check, then loses its parent, as if removed concurrently.
Node removed = mock(Node.class);
when(removed.getLevel()).thenReturn(dataNodes[1].getLevel());
when(removed.getAncestor(dataNodes[1].getLevel() - 1)).thenReturn(cluster.getNode(ROOT));
when(removed.getNetworkFullPath()).thenReturn(dataNodes[1].getNetworkFullPath());

assertEquals(2, cluster.getDistanceCost(dataNodes[0], dataNodes[1]));
assertEquals(Integer.MAX_VALUE, cluster.getDistanceCost(dataNodes[0], removed));
}

@ParameterizedTest
@MethodSource("topologies")
void testSortByDistanceCostWhileTopologyChanges(NodeSchema[] schemas, Node[] nodeArray) throws Exception {
initNetworkTopology(schemas, nodeArray);
List<Node> allNodes = Arrays.asList(dataNodes);
List<Node> stableNodes = new ArrayList<>();
List<Node> changingNodes = new ArrayList<>();
for (int i = 0; i < dataNodes.length; i++) {
(i % 2 == 0 ? stableNodes : changingNodes).add(dataNodes[i]);
}
Node reader = stableNodes.get(0);
AtomicBoolean done = new AtomicBoolean();
ExecutorService executor = Executors.newFixedThreadPool(5);
try {
Future<?> writer = executor.submit(() -> {
while (!done.get()) {
changingNodes.forEach(cluster::remove);
changingNodes.forEach(cluster::add);
}
});
List<Future<?>> sorters = new ArrayList<>();
for (int t = 0; t < 4; t++) {
sorters.add(executor.submit(() -> {
for (int i = 0; i < 2000; i++) {
assertThat(cluster.sortByDistanceCost(reader, allNodes, allNodes.size()))
.containsExactlyInAnyOrderElementsOf(allNodes);
List<? extends Node> sorted = cluster.sortByDistanceCost(reader, stableNodes, stableNodes.size());
for (int j = 0; j + 1 < sorted.size(); j++) {
assertThat(cluster.getDistanceCost(reader, sorted.get(j)))
.isLessThanOrEqualTo(cluster.getDistanceCost(reader, sorted.get(j + 1)));
}
}
}));
}
for (Future<?> sorter : sorters) {
sorter.get(1, TimeUnit.MINUTES);
}
done.set(true);
writer.get(1, TimeUnit.MINUTES);
} finally {
done.set(true);
executor.shutdownNow();
}
}

@Test
void testSingleNodeRackWithAffinityNode() {
// network topology with default cost
Expand Down
53 changes: 53 additions & 0 deletions hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,15 @@

import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY;
import static org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_NETWORK_TOPOLOGY_SCHEMA_FILE;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_NETWORK_TOPOLOGY_SCHEMA_FILE_DEFAULT;
import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT;
import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT_LEVEL;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.mockito.Mockito.mock;

Expand All @@ -36,6 +42,10 @@
import org.apache.hadoop.hdds.scm.HddsTestUtils;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub;
import org.apache.hadoop.hdds.scm.net.InnerNode;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
import org.apache.hadoop.hdds.scm.net.Node;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationInternalInterface;
import org.apache.hadoop.hdds.scm.server.SCMConfigurator;
Expand All @@ -45,6 +55,7 @@
import org.apache.hadoop.ozone.om.KeyManagerImpl;
import org.apache.hadoop.ozone.om.OmTestManagers;
import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.ozone.test.GenericTestUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -79,6 +90,8 @@ public static void setup() throws Exception {
config.set(NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
StaticMapping.class.getName());
config.set(OzoneConfigKeys.OZONE_NETWORK_TOPOLOGY_AWARE_READ_KEY, "true");
// Also the expiry of OM's resolved client nodes.
config.set(OzoneConfigKeys.OZONE_OM_NETWORK_TOPOLOGY_REFRESH_DURATION, "2s");
List<DatanodeDetails> datanodes = new ArrayList<>(NODE_COUNT);
List<String> nodeMapping = new ArrayList<>(NODE_COUNT);
for (int i = 0; i < NODE_COUNT; i++) {
Expand Down Expand Up @@ -229,6 +242,46 @@ public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() {
"Write pipeline order must be preserved when client is unresolved");
}

@Test
public void cachesResolvedClientPerClusterMap() {
NetworkTopology clusterMap = om.getClusterMap();
Node client = keyManager.getOtherNode("edge0", clusterMap);
assertNotNull(client);
assertEquals("/rack0", client.getNetworkLocation());
assertSame(client, keyManager.getOtherNode("edge0", clusterMap));

Node refreshedClient = keyManager.getOtherNode("edge0", newClusterMapInstance(clusterMap));
assertNotSame(client, refreshedClient);
assertEquals(client.getNetworkFullPath(), refreshedClient.getNetworkFullPath());
}

@Test
public void cachesClientOnRackWithoutDatanodes() {
NetworkTopology clusterMap = om.getClusterMap();
String client = "edge-on-rack-without-datanodes";
StaticMapping.addNodeToRack(client, "/rack-without-datanodes");
assertNull(keyManager.getOtherNode(client, clusterMap));

StaticMapping.addNodeToRack(client, "/rack0");
assertNull(keyManager.getOtherNode(client, clusterMap));
assertEquals("/rack0", keyManager.getOtherNode(client, newClusterMapInstance(clusterMap)).getNetworkLocation());
}

@Test
public void cachedClientExpiresAfterTopologyRefreshInterval() throws Exception {
NetworkTopology clusterMap = om.getClusterMap();
Node client = keyManager.getOtherNode("edge1", clusterMap);
assertNotNull(client);

GenericTestUtils.waitFor(() -> keyManager.getOtherNode("edge1", clusterMap) != client, 100, 10_000);
}

private NetworkTopology newClusterMapInstance(NetworkTopology clusterMap) {
return new NetworkTopologyImpl(
config.get(OZONE_SCM_NETWORK_TOPOLOGY_SCHEMA_FILE, OZONE_SCM_NETWORK_TOPOLOGY_SCHEMA_FILE_DEFAULT),
(InnerNode) clusterMap.getNode(ROOT));
}

private String nodeAddress(DatanodeDetails dn) {
boolean useHostname = config.getBoolean(
HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME,
Expand Down
61 changes: 51 additions & 10 deletions hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,8 @@
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Maps;
import jakarta.annotation.Nonnull;
import java.io.IOException;
Expand All @@ -114,6 +116,7 @@
import java.util.Stack;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
Expand All @@ -129,6 +132,7 @@
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.client.ScmTopologyClient;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.scm.net.InnerNode;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
Expand Down Expand Up @@ -227,6 +231,8 @@ public class KeyManagerImpl implements KeyManager {
private BackgroundService openKeyCleanupService;
private BackgroundService multipartUploadCleanupService;
private DNSToSwitchMapping dnsToSwitchMapping;
private final AtomicReference<ClientNodeCache> clientNodeCache = new AtomicReference<>();
private final Duration clientNodeCacheExpiry;
private CompactionService compactionService;
private KeyLifecycleService keyLifecycleService;

Expand Down Expand Up @@ -254,6 +260,7 @@ public KeyManagerImpl(OzoneManager om, ScmClient scmClient,
this.secretManager = secretManager;
this.kmsProvider = kmsProvider;
this.metrics = metrics;
this.clientNodeCacheExpiry = ScmTopologyClient.parseRefreshDuration(conf);
}

@Override
Expand Down Expand Up @@ -2381,23 +2388,57 @@ private Node getClientNode(String clientMachine,
return getOtherNode(clientMachine, clusterMap);
}

private Node getOtherNode(String clientMachine,
NetworkTopology clusterMap) {
@VisibleForTesting
public Node getOtherNode(String clientMachine,
NetworkTopology clusterMap) {
ClientNodeCache cache = clientNodeCache.get();
if (cache == null || cache.clusterMap != clusterMap) {
cache = new ClientNodeCache(clusterMap, clientNodeCacheExpiry);
clientNodeCache.set(cache);
}
Optional<Node> cached = cache.nodes.getIfPresent(clientMachine);
if (cached != null) {
return cached.orElse(null);
}
try {
String clientLocation = resolveNodeLocation(clientMachine);
if (clientLocation != null) {
Node rack = clusterMap.getNode(clientLocation);
if (rack instanceof InnerNode) {
return new NodeImpl(clientMachine, clientLocation,
(InnerNode) rack, rack.getLevel() + 1,
NODE_COST_DEFAULT);
}
if (clientLocation == null) {
return null;
}
// A client whose rack is not in this map is cached too: that only changes with the map.
Node rack = clusterMap.getNode(clientLocation);
Optional<Node> node = rack instanceof InnerNode
? Optional.of(new NodeImpl(clientMachine, clientLocation, (InnerNode) rack, rack.getLevel() + 1,
NODE_COST_DEFAULT))
: Optional.empty();
cache.nodes.put(clientMachine, node);
return node.orElse(null);
} catch (Exception e) {
LOG.info("Could not resolve client {}: {}",
clientMachine, e.getMessage());
return null;
}
}

/**
* Non-datanode clients resolved against one cluster map. The client node is linked to that map's
* rack, and OM replaces the map instance on refresh, so the cache is replaced along with it.
* Entries also expire after the topology refresh interval, which bounds staleness for mappings
* that do not cache their own results.
*/
private static final class ClientNodeCache {
private static final int MAX_SIZE = 10_000;

private final NetworkTopology clusterMap;
private final Cache<String, Optional<Node>> nodes;

private ClientNodeCache(NetworkTopology clusterMap, Duration expiry) {
this.clusterMap = clusterMap;
this.nodes = CacheBuilder.newBuilder()
.maximumSize(MAX_SIZE)
.expireAfterWrite(expiry.toMillis(), TimeUnit.MILLISECONDS)
.build();
}
return null;
}

private String resolveNodeLocation(String hostname) {
Expand Down