diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/net/NetworkTopologyImpl.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/net/NetworkTopologyImpl.java index f83b8fae1203..2510c70ac0bb 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/net/NetworkTopologyImpl.java +++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/net/NetworkTopologyImpl.java @@ -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; } @@ -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()); } /** @@ -796,7 +818,7 @@ public List 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> tree = new TreeMap<>(); diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/net/TestNetworkTopologyImpl.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/net/TestNetworkTopologyImpl.java index c71986cbb26d..62814a5be67b 100644 --- a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/net/TestNetworkTopologyImpl.java +++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/net/TestNetworkTopologyImpl.java @@ -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; @@ -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; @@ -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 nodeList = Arrays.asList(dataNodes); + + List 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 allNodes = Arrays.asList(dataNodes); + List stableNodes = new ArrayList<>(); + List 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> 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 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 diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java index 6ff81b327bcf..f3260c24a520 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java @@ -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; @@ -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; @@ -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; @@ -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 datanodes = new ArrayList<>(NODE_COUNT); List nodeMapping = new ArrayList<>(NODE_COUNT); for (int i = 0; i < NODE_COUNT; i++) { @@ -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, diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index 445160972d82..64aac36f8cfe 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java @@ -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; @@ -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; @@ -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; @@ -227,6 +231,8 @@ public class KeyManagerImpl implements KeyManager { private BackgroundService openKeyCleanupService; private BackgroundService multipartUploadCleanupService; private DNSToSwitchMapping dnsToSwitchMapping; + private final AtomicReference clientNodeCache = new AtomicReference<>(); + private final Duration clientNodeCacheExpiry; private CompactionService compactionService; private KeyLifecycleService keyLifecycleService; @@ -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 @@ -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 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 = 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> 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) {