jojochuang commented on code in PR #10693:
URL: https://github.com/apache/ozone/pull/10693#discussion_r3648257849
##########
hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java:
##########
@@ -1052,4 +1070,439 @@ private void scheduleTasks(
rs.addTask(new ReplicationTask(toTarget(i, target), noopReplicator));
}
}
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumeDisabledUsesGlobalPool(ContainerLayoutVersion layout) {
+ this.layoutVersion = layout;
+ ReplicationServer.ReplicationConfig repConf =
+ new ReplicationServer.ReplicationConfig();
+ repConf.setPerVolumeEnabled(false);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ assertNull(supervisor.getVolumeReplicationThreadPools());
+ replicatorRef.set(doneReplicator);
+ supervisor.addTask(createTask(1L));
+ assertEquals(1, supervisor.getReplicationSuccessCount());
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumeInitLogging(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ LogCapturer supervisorLogs =
+ LogCapturer.captureLogs(ReplicationSupervisor.class);
+ LogCapturer poolLogs =
+ LogCapturer.captureLogs(VolumeReplicationThreadPools.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ assertNotNull(supervisor.getVolumeReplicationThreadPools());
+ assertThat(supervisorLogs.getOutput())
+ .contains("Per-volume container replication thread pools enabled");
+ assertThat(poolLogs.getOutput())
+ .contains("Initialized 2 per-volume replication thread pools");
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertThat(poolLogs.getOutput())
+ .contains(volume.getStorageDir().getPath());
+ }
+ } finally {
+ supervisorLogs.stopCapturing();
+ poolLogs.stopCapturing();
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolSizeRespected(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 3);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(3, pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolResize(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(newDirectExecutorService())
+ .clock(clock)
+ .build();
+
+ try {
+ supervisor.setPerVolumePoolSize(3);
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(3, pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ assertEquals(3, repConf.getPerVolumeStreamsLimit());
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolResizeOnNodeStateChange(ContainerLayoutVersion
layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 2);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ try {
+ datanode.setPersistedOpState(IN_SERVICE);
+ supervisor.nodeStateUpdated(
+ HddsProtos.NodeOperationalState.DECOMMISSIONING);
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ int expected = repConf.scaleOutOfServiceLimit(2);
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(expected,
+ pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePoolResizeDuringDecommission(ContainerLayoutVersion
layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 2);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .clock(clock)
+ .build();
+
+ try {
+ datanode.setPersistedOpState(IN_SERVICE);
+ supervisor.nodeStateUpdated(DECOMMISSIONING);
+ supervisor.setPerVolumePoolSize(2);
+ VolumeReplicationThreadPools pools =
+ supervisor.getVolumeReplicationThreadPools();
+ int expected = repConf.scaleOutOfServiceLimit(2);
+ for (StorageVolume volume : volumeSet.getVolumesList()) {
+ assertEquals(expected,
+ pools.getPoolSize(volume.getStorageDir().getPath()));
+ }
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void nonPushReplicationUsesGlobalPoolWhenPerVolumeEnabled(
+ ContainerLayoutVersion layout, @TempDir File perVolumeTempDir) throws
Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+ AtomicInteger globalExecutions = new AtomicInteger();
+
+ ExecutorService trackingGlobal = new AbstractExecutorService() {
+ @Override
+ public void shutdown() {
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return emptyList();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return false;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return false;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) {
+ return true;
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ globalExecutions.incrementAndGet();
+ command.run();
+ }
+ };
+
+ ReplicationSupervisor supervisor = ReplicationSupervisor.newBuilder()
+ .stateContext(context)
+ .replicationConfig(repConf)
+ .containerSet(set)
+ .volumeSet(volumeSet)
+ .executor(trackingGlobal)
+ .clock(clock)
+ .build();
+
+ try {
+ supervisor.addTask(createReconciliationTask(1L));
+ assertEquals(1, globalExecutions.get());
+ assertEquals(1, supervisor.getReplicationSuccessCount());
+ } finally {
+ supervisor.stop();
+ }
+ }
+
+ @ContainerLayoutTestInfo.ContainerTest
+ public void perVolumePushIsolation(ContainerLayoutVersion layout,
+ @TempDir File perVolumeTempDir) throws Exception {
+ this.layoutVersion = layout;
+ OzoneConfiguration conf = perVolumeConf(perVolumeTempDir, 1);
+ MutableVolumeSet volumeSet = newVolumeSet(conf);
+ HddsVolume vol1 = (HddsVolume) volumeSet.getVolumesList().get(0);
+ HddsVolume vol2 = (HddsVolume) volumeSet.getVolumesList().get(1);
+
+ addContainerOnVolume(1L, vol1, conf);
+ addContainerOnVolume(2L, vol2, conf);
+
+ ReplicationServer.ReplicationConfig repConf =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+
+ CountDownLatch vol1Started = new CountDownLatch(1);
+ CountDownLatch vol1Release = new CountDownLatch(1);
+ ContainerReplicator volumeAwareReplicator = task -> {
+ Container<?> container = set.getContainer(task.getContainerId());
+ HddsVolume volume = container.getContainerData().getVolume();
+ if (volume == vol1) {
+ vol1Started.countDown();
+ assertDoesNotThrow(() -> vol1Release.await(10, TimeUnit.SECONDS));
+ }
Review Comment:
Fixed in 28798f: assertTrue(await…) for vol1Release / task1Block (with
interrupt handling in the replicator lambda).
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/VolumeReplicationThreadPools.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.replication;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.hdds.utils.HddsServerUtil;
+import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Per-volume replication handler thread pools for push-based replication.
+ */
+final class VolumeReplicationThreadPools {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(VolumeReplicationThreadPools.class);
+
+ private final ConcurrentHashMap<String, ThreadPoolExecutor> pools =
+ new ConcurrentHashMap<>();
+ private int currentPoolSize;
+
+ void init(Collection<? extends StorageVolume> volumes, int poolSize,
+ String threadNamePrefix) {
+ currentPoolSize = poolSize;
+ List<String> volumeRoots = new ArrayList<>();
+ for (StorageVolume volume : volumes) {
+ File storageDir = volume.getStorageDir();
+ String volumeRoot = storageDir.getPath();
+ volumeRoots.add(volumeRoot);
+ pools.put(volumeRoot,
+ createPool(poolSize, threadNamePrefix, storageDir.getName()));
+ }
Review Comment:
Fixed in 28798f: per-volume pool threads use full volumeRoot in the name via
a custom ThreadFactory.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java:
##########
@@ -318,7 +320,9 @@ public String getNamespace() {
.register(OZONE_BLOCK_DELETING_SERVICE_TIMEOUT,
this::reconfigBlockDeletingServiceTimeout)
.register(REPLICATION_STREAMS_LIMIT_KEY,
- this::reconfigReplicationStreamsLimit);
+ this::reconfigReplicationStreamsLimit)
+ .register(PER_VOLUME_STREAMS_LIMIT_KEY,
Review Comment:
Addressed in 28798f: @Config text for streams.limit vs
per.volume.streams.limit and Decommission.md / Reconfigurability.md.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java:
##########
@@ -719,6 +723,28 @@ private String reconfigReplicationStreamsLimit(String
value) {
return value;
}
+ private String reconfigPerVolumeStreamsLimit(String value) {
+ int newSize = Integer.parseInt(value);
+ Preconditions.checkArgument(newSize >= 1,
+ PER_VOLUME_STREAMS_LIMIT_KEY + " must be at least 1 but was %s",
+ value);
+ ReplicationConfig replicationConfig =
+ getDatanodeStateMachine().getSupervisor().getReplicationConfig();
+ if (!replicationConfig.isPerVolumeEnabled()) {
Review Comment:
Yes — staging per.volume.streams.limit while disabled is intentional;
description calls out it applies when per.volume.enabled is true.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationServer.java:
##########
@@ -224,6 +230,29 @@ public static final class ReplicationConfig {
)
private double outOfServiceFactor = OUTOFSERVICE_FACTOR_DEFAULT;
+ @Config(key = PER_VOLUME_ENABLED_KEY,
+ type = ConfigType.BOOLEAN,
+ defaultValue = "false",
+ tags = {DATANODE},
+ description = "When true, push-based container replication uses a " +
+ "separate replication handler thread pool per data volume so " +
+ "that slow replication on one disk does not block replication " +
+ "on other disks. Pull replication and other replication tasks " +
+ "continue to use the global replication handler thread pool."
+ )
+ private boolean perVolumeEnabled = false;
+
+ @Config(key = PER_VOLUME_STREAMS_LIMIT_KEY,
+ type = ConfigType.INT,
+ defaultValue = "1",
+ reconfigurable = true,
+ tags = {DATANODE},
+ description = "The maximum number of concurrent push replication " +
Review Comment:
Documented in 28798f: when per-volume is enabled, push uses per-volume
pools; streams.limit applies to the global pool (non-push + fallback push).
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -371,6 +501,20 @@ private void resize(HddsProtos.NodeOperationalState
nodeState) {
maxQueueSize = newMaxQueueSize;
executorThreadUpdater.accept(threadCount);
+
+ if (volumePools != null) {
Review Comment:
28798f clarifies maxReplicationStreams is global executor max only. Separate
Jira to follow for aggregate / per-volume capacity gauges.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
