jojochuang commented on code in PR #10122:
URL: https://github.com/apache/ozone/pull/10122#discussion_r3599327168
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/UnhealthyReplicationProcessor.java:
##########
@@ -105,6 +112,12 @@ public void processAll(ReplicationQueue queue) {
.getMetrics().incrPendingReplicationLimitReachedTotal();
break;
}
+ if (reconstructionLimitReached(replicationManager)) {
+ LOG.info("The maximum number of pending reconstruction commands ({}) "
+
+ "are scheduled. Ending the iteration.",
+ replicationManager.getReconstructionInFlightLimit());
+ break;
+ }
Review Comment:
Addressed in 61e4eac3650: enforcement was moved from the processor loop to
`sendThrottledReconstructionCommand()` so 1-1 replication continues when the
reconstruction cap is reached. The processor no longer breaks on reconstruction
limit.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -537,6 +581,12 @@ public void sendThrottledReplicationCommand(ContainerInfo
containerInfo,
public void sendThrottledReconstructionCommand(ContainerInfo containerInfo,
ReconstructECContainersCommand command)
throws CommandTargetOverloadedException, NotLeaderException {
+ if (isReconstructionLimitReached()) {
+ metrics.incrECReconstructionCmdsDeferredTotal();
+ throw new CommandTargetOverloadedException(
+ "Global reconstruction limit (" + getReconstructionInFlightLimit()
+ + ") reached for container " + containerInfo.getContainerID());
+ }
Review Comment:
Fixed in c196dd48a0f: `tryReserveReconstructionSlot()` atomically reserves a
slot via CAS before sending the command. The increment is no longer deferred to
`adjustPendingOpsAndMetrics()`.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1802,186 @@ private void mockReplicationCommandCounts(
});
}
+ @Test
+ public void testReconstructionGlobalLimitDisabledByDefault() throws
IOException {
+ assertEquals(0, rmConf.getReconstructionGlobalLimit());
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(0, rm.getReconstructionInFlightLimit());
+ assertFalse(rm.isReconstructionLimitReached());
+ }
+
+ @Test
+ public void testInflightReconstructionLimit() throws IOException,
NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(2, rm.getReconstructionInFlightLimit());
+ assertEquals(0, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ // Send one reconstruction command with 2 fragments
Review Comment:
Fixed in c196dd48a0f: trailing whitespace removed from the test.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -422,6 +435,57 @@ public long getInflightReplicationCount() {
.getPendingOpCount(ContainerReplicaOp.PendingOpType.ADD);
}
+ /**
+ * Returns the number of active EC reconstruction commands currently in
+ * progress across the cluster.
+ */
+ public int getInflightReconstructionCount() {
+ return inflightReconstructionCount.get();
+ }
+
+ /**
+ * Returns the maximum number of inflight reconstruction commands allowed
+ * across the cluster at any given time.
+ * @return the maximum number of inflight reconstruction commands allowed
+ */
+ public int getReconstructionInFlightLimit() {
+ return rmConf.getReconstructionGlobalLimit();
+ }
+
+ /**
+ * Returns true if the number of inflight reconstruction commands has reached
+ * the global limit.
+ * @return true if the limit is reached, false otherwise
+ */
+ public boolean isReconstructionLimitReached() {
+ int limit = getReconstructionInFlightLimit();
+ return limit > 0 && getInflightReconstructionCount() >= limit;
+ }
+
+ private void decrementInflightReconstructionCount() {
+ inflightReconstructionCount.updateAndGet(count -> Math.max(0, count - 1));
+ }
+
+ private boolean tryReserveReconstructionSlot() {
+ int limit = getReconstructionInFlightLimit();
+ if (limit <= 0) {
+ return true;
+ }
+ while (true) {
+ int current = inflightReconstructionCount.get();
+ if (current >= limit) {
+ return false;
+ }
+ if (inflightReconstructionCount.compareAndSet(current, current + 1)) {
+ return true;
+ }
+ }
+ }
Review Comment:
Fixed in aad8da6d69d: `tryReserveReconstructionSlot()` always increments the
counter; the global limit is only enforced when `reconstruction.global.limit >
0`.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -537,17 +601,31 @@ public void sendThrottledReplicationCommand(ContainerInfo
containerInfo,
public void sendThrottledReconstructionCommand(ContainerInfo containerInfo,
ReconstructECContainersCommand command)
throws CommandTargetOverloadedException, NotLeaderException {
- List<DatanodeDetails> targets = command.getTargetDatanodes();
- List<Pair<Integer, DatanodeDetails>> targetWithCmds =
- getAvailableDatanodesForReplication(targets);
- if (targetWithCmds.isEmpty()) {
+ if (!tryReserveReconstructionSlot()) {
metrics.incrECReconstructionCmdsDeferredTotal();
- throw new CommandTargetOverloadedException("No target with capacity " +
- "available for reconstruction of " + containerInfo.getContainerID());
+ throw new CommandTargetOverloadedException(
+ "Global reconstruction limit (" + getReconstructionInFlightLimit()
+ + ") reached for container " + containerInfo.getContainerID());
Review Comment:
PR description updated to document send-path enforcement in
`sendThrottledReconstructionCommand()` rather than processor dequeue.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ private void mockReplicationCommandCounts(
});
}
+ @Test
+ public void testReconstructionGlobalLimitDisabledByDefault() throws
IOException {
+ assertEquals(0, rmConf.getReconstructionGlobalLimit());
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(0, rm.getReconstructionInFlightLimit());
+ assertFalse(rm.isReconstructionLimitReached());
+ }
+
+ @Test
+ public void testInflightReconstructionLimit() throws IOException,
NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(2, rm.getReconstructionInFlightLimit());
+ assertEquals(0, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ // Send one reconstruction command with 2 fragments
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)),
+ (ECReplicationConfig) repConfig);
+
+ rm.sendThrottledReconstructionCommand(container, cmd1);
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Send another reconstruction command with 1 fragment
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)),
+ (ECReplicationConfig) repConfig);
+ rm.sendThrottledReconstructionCommand(container, cmd2);
+ assertEquals(2, rm.getInflightReconstructionCount());
+ assertTrue(rm.isReconstructionLimitReached());
+
+ // Complete one fragment of cmd1
+ ContainerReplicaOp op1 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(0), 1, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op1, container.containerID(), false);
+ // Still 2 because cmd1 is not fully finished
+ assertEquals(2, rm.getInflightReconstructionCount());
+
+ // Complete second fragment of cmd1
+ ContainerReplicaOp op2 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(1), 2, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op2, container.containerID(), false);
+ // Now 1
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Complete cmd2
+ ContainerReplicaOp op3 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd2.getTargetDatanodes().get(0), 3, cmd2, Long.MAX_VALUE, 0);
+ rm.opCompleted(op3, container.containerID(), false);
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testSendReconstructionCommandRejectedWhenGlobalLimitReached()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
(ECReplicationConfig) repConfig);
Review Comment:
Fixed in aad8da6d69d: wrapped long constructor arguments onto separate lines.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ private void mockReplicationCommandCounts(
});
}
+ @Test
+ public void testReconstructionGlobalLimitDisabledByDefault() throws
IOException {
+ assertEquals(0, rmConf.getReconstructionGlobalLimit());
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(0, rm.getReconstructionInFlightLimit());
+ assertFalse(rm.isReconstructionLimitReached());
+ }
+
+ @Test
+ public void testInflightReconstructionLimit() throws IOException,
NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(2, rm.getReconstructionInFlightLimit());
+ assertEquals(0, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ // Send one reconstruction command with 2 fragments
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)),
+ (ECReplicationConfig) repConfig);
+
+ rm.sendThrottledReconstructionCommand(container, cmd1);
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Send another reconstruction command with 1 fragment
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)),
+ (ECReplicationConfig) repConfig);
+ rm.sendThrottledReconstructionCommand(container, cmd2);
+ assertEquals(2, rm.getInflightReconstructionCount());
+ assertTrue(rm.isReconstructionLimitReached());
+
+ // Complete one fragment of cmd1
+ ContainerReplicaOp op1 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(0), 1, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op1, container.containerID(), false);
+ // Still 2 because cmd1 is not fully finished
+ assertEquals(2, rm.getInflightReconstructionCount());
+
+ // Complete second fragment of cmd1
+ ContainerReplicaOp op2 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(1), 2, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op2, container.containerID(), false);
+ // Now 1
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Complete cmd2
+ ContainerReplicaOp op3 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd2.getTargetDatanodes().get(0), 3, cmd2, Long.MAX_VALUE, 0);
+ rm.opCompleted(op3, container.containerID(), false);
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testSendReconstructionCommandRejectedWhenGlobalLimitReached()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
(ECReplicationConfig) repConfig);
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(2)),
(ECReplicationConfig) repConfig);
Review Comment:
Fixed in aad8da6d69d: wrapped long constructor arguments onto separate lines.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java:
##########
@@ -1801,4 +1807,240 @@ private void mockReplicationCommandCounts(
});
}
+ @Test
+ public void testReconstructionGlobalLimitDisabledByDefault() throws
IOException {
+ assertEquals(0, rmConf.getReconstructionGlobalLimit());
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(0, rm.getReconstructionInFlightLimit());
+ assertFalse(rm.isReconstructionLimitReached());
+ }
+
+ @Test
+ public void testInflightReconstructionLimit() throws IOException,
NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ assertEquals(2, rm.getReconstructionInFlightLimit());
+ assertEquals(0, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ // Send one reconstruction command with 2 fragments
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)),
+ (ECReplicationConfig) repConfig);
+
+ rm.sendThrottledReconstructionCommand(container, cmd1);
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Send another reconstruction command with 1 fragment
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)),
+ (ECReplicationConfig) repConfig);
+ rm.sendThrottledReconstructionCommand(container, cmd2);
+ assertEquals(2, rm.getInflightReconstructionCount());
+ assertTrue(rm.isReconstructionLimitReached());
+
+ // Complete one fragment of cmd1
+ ContainerReplicaOp op1 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(0), 1, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op1, container.containerID(), false);
+ // Still 2 because cmd1 is not fully finished
+ assertEquals(2, rm.getInflightReconstructionCount());
+
+ // Complete second fragment of cmd1
+ ContainerReplicaOp op2 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd1.getTargetDatanodes().get(1), 2, cmd1, Long.MAX_VALUE, 0);
+ rm.opCompleted(op2, container.containerID(), false);
+ // Now 1
+ assertEquals(1, rm.getInflightReconstructionCount());
+ assertFalse(rm.isReconstructionLimitReached());
+
+ // Complete cmd2
+ ContainerReplicaOp op3 = new ContainerReplicaOp(
+ ContainerReplicaOp.PendingOpType.ADD,
+ cmd2.getTargetDatanodes().get(0), 3, cmd2, Long.MAX_VALUE, 0);
+ rm.opCompleted(op3, container.containerID(), false);
+ assertEquals(0, rm.getInflightReconstructionCount());
+ }
+
+ @Test
+ public void testSendReconstructionCommandRejectedWhenGlobalLimitReached()
+ throws IOException, NodeNotFoundException {
+ rmConf.setReconstructionGlobalLimit(2);
+ ReplicationManager rm = createReplicationManager();
+ mockReplicationCommandCounts(dn -> 0, dn -> 0);
+
+ ContainerInfo container = ReplicationTestUtil.createContainerInfo(
+ repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20);
+
+ ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand(
+ 1L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)),
(ECReplicationConfig) repConfig);
+ ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand(
+ 2L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(2)),
(ECReplicationConfig) repConfig);
+ ReconstructECContainersCommand cmd3 = new ReconstructECContainersCommand(
+ 3L, Collections.emptyList(),
+ ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()),
+ ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)),
(ECReplicationConfig) repConfig);
Review Comment:
Fixed in aad8da6d69d: wrapped long constructor arguments onto separate lines.
--
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]
