smengcl commented on code in PR #11651:
URL: https://github.com/apache/ozone/pull/11651#discussion_r4332761247
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java:
##########
@@ -656,48 +677,83 @@ public synchronized SubmitSnapshotDiffResponse
submitSnapshotDiff(
SnapshotDiffJob snapDiffJob = getSnapDiffReportStatus(snapDiffJobKey,
volumeName, bucketName, fromSnapshotName, toSnapshotName,
- forceFullDiff, disableNativeDiff);
+ forceFullDiff, disableNativeDiff, orderedReport);
JobStatus prevStatus = snapDiffJob.getStatus();
String prevReason = isEmpty(snapDiffJob.getReason()) ? null :
snapDiffJob.getReason();
String jobId = snapDiffJob.getJobId();
+ boolean createNewJob = false;
+ boolean submitJob = false;
+ boolean isFso = getBucketLayout(volumeName, bucketName,
activeOmMetadataManager).isFileSystemOptimized();
+ String responseMsg = "";
+
if (prevStatus == FAILED || prevStatus == REJECTED || prevStatus ==
CANCELLED) {
+ createNewJob = true;
+ submitJob = true;
+ } else if ((prevStatus == QUEUED || prevStatus == IN_PROGRESS ||
prevStatus == DONE) &&
Review Comment:
This also supersedes a job that is still IN_PROGRESS, but nothing stops the
old task:
- `areDiffJobAndSnapshotsActive` only checks the status under `jobKey`, and
the new job is IN_PROGRESS too, so the old task runs to completion and keeps an
executor slot.
- `recordActivity` and `updateProgress` are keyed by `jobKey` with no jobId
check, so the old task overwrites the new job's subStatus and progress.
- `createSnapDiffJob` enqueues the old jobId for purge right away. If
`SnapshotDiffCleanupService#removeOlderJobReport` runs before the old task
writes its report, the rows it writes afterwards are never cleaned up.
Simplest is to only supersede a QUEUED or DONE job; an IN_PROGRESS one then
falls through to the usual "get the report in N ms" response, and the next
`--ordered` submit after it is DONE reruns it. A test resubmitting `--ordered`
against an IN_PROGRESS job would be good too.
```suggestion
} else if ((prevStatus == QUEUED || prevStatus == DONE) &&
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/MergeJoinSnapDiffWriter.java:
##########
@@ -121,11 +147,37 @@ private static void classifyMergeJoin(SnapDiffJobStore
store, boolean isFso) thr
newEntry = newHead.hasNext() ? newHead.next() : null;
oldEntry = oldHead.hasNext() ? oldHead.next() : null;
}
+ mergeSteps++;
+ SnapDiffProgressHelper.reportSlice(reportProgress, 0.0, classifyEnd,
Review Comment:
`reportProgress` ends up in `SnapshotDiffManager#updateProgress`, which is a
synchronized get, serialize and put on `snapDiffJobTable`. Calling it on every
merge step means one RocksDB write per entry, each taking the manager lock that
submit and get report also need. Could this use `TenPercentCheckpoint` like the
scan readers?
```diff
long mergeSteps = 0L;
long mergeDenominator = store.mergeJoinStepDenominator();
double classifyEnd = isFso ? FSO_CLASSIFY_END : 1.0;
+ SnapDiffProgressHelper.TenPercentCheckpoint checkpoint = new
SnapDiffProgressHelper.TenPercentCheckpoint();
...
mergeSteps++;
- SnapDiffProgressHelper.reportSlice(reportProgress, 0.0, classifyEnd,
- (double) mergeSteps / mergeDenominator);
+ SnapDiffProgressHelper.reportEveryTenPercentInSlice(reportProgress,
0.0, classifyEnd, mergeSteps,
+ mergeDenominator, checkpoint);
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/MergeJoinSnapDiffWriter.java:
##########
@@ -121,11 +147,37 @@ private static void classifyMergeJoin(SnapDiffJobStore
store, boolean isFso) thr
newEntry = newHead.hasNext() ? newHead.next() : null;
oldEntry = oldHead.hasNext() ? oldHead.next() : null;
}
+ mergeSteps++;
+ SnapDiffProgressHelper.reportSlice(reportProgress, 0.0, classifyEnd,
+ (double) mergeSteps / mergeDenominator);
}
} catch (RocksDatabaseException e) {
throw new IOException(e);
}
store.flushWrites();
+ if (reportProgress != null) {
+ reportProgress.accept(isFso ? FSO_CLASSIFY_END : 1.0);
+ }
+ }
+
+ private static final class ResolveProgress {
+ private final DoubleConsumer reportProgress;
+ private final long resolveDenominator;
+ private final double resolveEnd;
+ private long resolveSteps;
+
+ ResolveProgress(SnapDiffJobStore store, DoubleConsumer reportProgress,
+ boolean dependencyOrderingEnabled) {
+ this.reportProgress = reportProgress;
+ this.resolveDenominator = store.classifiedEntryDenominator();
+ this.resolveEnd = dependencyOrderingEnabled ? FSO_RESOLVE_END_ORDERED :
FSO_RESOLVE_END_UNORDERED;
+ }
+
+ void onClassifiedEntry(DiffType diffType) {
+ resolveSteps++;
+ SnapDiffProgressHelper.reportSlice(reportProgress, FSO_CLASSIFY_END,
resolveEnd,
+ (double) resolveSteps / resolveDenominator);
Review Comment:
Same here, this writes job progress to RocksDB for every classified entry.
With a `TenPercentCheckpoint` field on `ResolveProgress`:
```suggestion
SnapDiffProgressHelper.reportEveryTenPercentInSlice(reportProgress,
FSO_CLASSIFY_END, resolveEnd,
resolveSteps, resolveDenominator, checkpoint);
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java:
##########
@@ -656,48 +677,83 @@ public synchronized SubmitSnapshotDiffResponse
submitSnapshotDiff(
SnapshotDiffJob snapDiffJob = getSnapDiffReportStatus(snapDiffJobKey,
volumeName, bucketName, fromSnapshotName, toSnapshotName,
- forceFullDiff, disableNativeDiff);
+ forceFullDiff, disableNativeDiff, orderedReport);
JobStatus prevStatus = snapDiffJob.getStatus();
String prevReason = isEmpty(snapDiffJob.getReason()) ? null :
snapDiffJob.getReason();
String jobId = snapDiffJob.getJobId();
+ boolean createNewJob = false;
+ boolean submitJob = false;
+ boolean isFso = getBucketLayout(volumeName, bucketName,
activeOmMetadataManager).isFileSystemOptimized();
+ String responseMsg = "";
+
if (prevStatus == FAILED || prevStatus == REJECTED || prevStatus ==
CANCELLED) {
+ createNewJob = true;
+ submitJob = true;
+ } else if ((prevStatus == QUEUED || prevStatus == IN_PROGRESS ||
prevStatus == DONE) &&
+ (isFso && orderedReport && !snapDiffJob.isOrderedReport())) {
+ createNewJob = true;
+ submitJob = true;
+ responseMsg = "Previous snapshot diff attempt found with an unordered
report. Status: " + prevStatus
+ + ".\nSubmitting a new job with ordered report. Please retry after "
+ defaultWaitTime +
+ " ms using --get-report option to get the report.\n";
+ } else if (prevStatus == QUEUED) {
+ submitJob = true;
+ }
+
+ if (createNewJob) {
jobId = createSnapDiffJob(snapDiffJobKey, volumeName, bucketName,
fromSnapshotName,
- toSnapshotName, forceFullDiff, disableNativeDiff, prevStatus,
snapDiffJob);
+ toSnapshotName, forceFullDiff, disableNativeDiff, orderedReport,
QUEUED, snapDiffJob);
}
final String submitJobId = jobId;
- switch (prevStatus) {
- case QUEUED:
- case FAILED:
- case REJECTED:
- case CANCELLED:
+ if (submitJob) {
LOG.info("Submitting snapshot diff generation request for" +
" volume: {}, bucket: {}, fromSnapshot: {} and toSnapshot: {}",
volumeName, bucketName, fromSnapshotName, toSnapshotName);
try {
- updateJobStatus(snapDiffJobKey, prevStatus, IN_PROGRESS);
+ updateJobStatus(snapDiffJobKey, QUEUED, IN_PROGRESS);
snapDiffExecutor.execute(() ->
generateSnapshotDiffReport(snapDiffJobKey, submitJobId,
volumeName, bucketName, fromSnapshotName, toSnapshotName,
- forceFullDiff, disableNativeDiff));
- return new SubmitSnapshotDiffResponse(defaultWaitTime, prevStatus,
prevReason);
+ forceFullDiff, disableNativeDiff, orderedReport));
} catch (RejectedExecutionException exception) {
updateJobStatus(snapDiffJobKey, IN_PROGRESS, REJECTED);
LOG.warn("Exceeded the snapDiff parallel requests processing " +
- "limit. Rejecting the snapDiff job: {}.", snapDiffJobKey);
- return new SubmitSnapshotDiffResponse("Snapshot diff job was rejected
because parallel processing of " +
- "snap diff jobs exceeded the limit. Please try again in " +
defaultWaitTime + " ms.");
+ "limit. Rejecting the snapDiff job: {}.", snapDiffJobKey);
+ responseMsg = "Snapshot diff job was rejected because parallel
processing of " +
+ "snap diff jobs exceeded the limit. Please try again in " +
defaultWaitTime + " ms.";
} catch (Exception exception) {
updateJobStatusToFailed(snapDiffJobKey, exception.getMessage());
LOG.error("Snapshot diff job: {} failed.", snapDiffJobKey, exception);
- return new SubmitSnapshotDiffResponse("Snapshot diff job submission
failed. Reason: " + exception.getMessage());
+ responseMsg = "Snapshot diff job submission failed. Reason: " +
exception.getMessage();
}
- case IN_PROGRESS:
- case DONE:
- return new SubmitSnapshotDiffResponse(defaultWaitTime, prevStatus,
prevReason);
- default:
- throw new IllegalStateException("Unknown snapshot job status: " +
snapDiffJob.getStatus());
}
+ if (!isEmpty(responseMsg)) {
+ return new SubmitSnapshotDiffResponse(responseMsg);
+ }
+ return new SubmitSnapshotDiffResponse(defaultWaitTime, prevStatus,
prevReason);
+ }
+
+ private SnapshotDiffResponse newDoneSnapshotDiffResponse(SnapshotDiffJob
snapDiffJob,
+ String volumeName, String bucketName, String fromSnapshotName, String
toSnapshotName,
+ String index, int pageSize) throws IOException {
+ SnapshotDiffReportOzone report = createPageResponse(snapDiffJob,
volumeName, bucketName,
+ fromSnapshotName, toSnapshotName, index, pageSize);
+ boolean dependencyOrdered = isDependencyOrderedReport(snapDiffJob,
volumeName, bucketName);
+ return new SnapshotDiffResponse(report, DONE, 0L,
dependencyOrderingLine(dependencyOrdered));
+ }
+
+ private boolean isDependencyOrderedReport(SnapshotDiffJob snapDiffJob,
String volumeName,
+ String bucketName) throws IOException {
+ if (getBucketLayout(volumeName, bucketName,
ozoneManager.getMetadataManager())
+ .isFileSystemOptimized()) {
+ return snapDiffJob.isOrderedReport();
Review Comment:
This returns the requested flag, not what was actually produced:
- With `ozone.om.snapshot.diff.optimized.enabled=false` (the default), or
the non native DAG fallback, `generateLegacySnapshotDiffReport` ignores
`orderedReport`. So `--ordered` gives an unordered report labelled "dependency
ordered", and on a DONE FSO job it also triggers a full rerun that produces the
same report.
- `MergeJoinSnapDiffWriter` switches to unordered output when the dependency
node count exceeds `maxInMemoryEntries` (`dependencyOrderingEnabled[0] =
false`), and the job still says ordered.
Could `writeReport` return whether ordering was actually applied
(`dependencyOrderingEnabledRef[0]`) and the job record that on DONE, with the
flag forced to false when the legacy path runs?
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java:
##########
@@ -93,6 +98,7 @@
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
+import java.util.function.DoubleConsumer;
Review Comment:
Import order, checkstyle will flag this. It belongs with the other
`java.util.function` imports:
```diff
import java.util.concurrent.Callable;
-import java.util.function.DoubleConsumer;
import java.util.concurrent.ExecutorService;
...
import java.util.function.BiFunction;
import java.util.function.Consumer;
+import java.util.function.DoubleConsumer;
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffJobStore.java:
##########
@@ -33,6 +33,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.EnumMap;
import java.util.Comparator;
Review Comment:
Import order:
```suggestion
import java.util.Comparator;
import java.util.EnumMap;
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java:
##########
@@ -136,8 +143,13 @@
import org.apache.hadoop.ozone.om.helpers.WithObjectID;
import org.apache.hadoop.ozone.om.helpers.WithParentObjectId;
import org.apache.hadoop.ozone.om.snapshot.db.SnapshotDiffDBDefinition;
+import org.apache.hadoop.ozone.om.snapshot.diff.DagDiffSequentialReader;
+import org.apache.hadoop.ozone.om.snapshot.diff.FullDiffSequentialReader;
+import org.apache.hadoop.ozone.om.snapshot.diff.MergeJoinSnapDiffWriter;
+import org.apache.hadoop.ozone.om.snapshot.diff.SnapDiffJobStore;
import
org.apache.hadoop.ozone.om.snapshot.diff.delta.CompositeDeltaDiffComputer;
import org.apache.hadoop.ozone.om.snapshot.diff.delta.DeltaFileComputer;
+import
org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager.ReadableOmSnapshotLocalDataProvider;
Review Comment:
Import order, this should come before `snapshot.db`:
```suggestion
import
org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager.ReadableOmSnapshotLocalDataProvider;
import org.apache.hadoop.ozone.om.snapshot.db.SnapshotDiffDBDefinition;
import org.apache.hadoop.ozone.om.snapshot.diff.DagDiffSequentialReader;
import org.apache.hadoop.ozone.om.snapshot.diff.FullDiffSequentialReader;
import org.apache.hadoop.ozone.om.snapshot.diff.MergeJoinSnapDiffWriter;
import org.apache.hadoop.ozone.om.snapshot.diff.SnapDiffJobStore;
import
org.apache.hadoop.ozone.om.snapshot.diff.delta.CompositeDeltaDiffComputer;
import org.apache.hadoop.ozone.om.snapshot.diff.delta.DeltaFileComputer;
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/MergeJoinSnapDiffWriter.java:
##########
@@ -27,6 +27,7 @@
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.function.DoubleConsumer;
Review Comment:
Import order, checkstyle will flag this:
```diff
import java.util.ArrayList;
-import java.util.function.DoubleConsumer;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.DoubleConsumer;
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/SnapDiffProgressHelper.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.om.snapshot.diff;
+
+import java.util.function.DoubleConsumer;
+
+/**
+ * Shared 10% checkpoint reporting for snapshot-diff scan stages.
+ */
+public final class SnapDiffProgressHelper {
+
+ private static final double TEN_PCT_STEP = 0.1;
+
+ private SnapDiffProgressHelper() {
+ }
+
+ static void reportSlice(DoubleConsumer progress, double sliceStart, double
sliceEnd,
+ double completedFraction) {
+ if (progress == null) {
+ return;
+ }
+ double width = sliceEnd - sliceStart;
+ progress.accept(sliceStart + width * Math.min(1.0, Math.max(0.0,
completedFraction)));
+ }
+
+ static void reportEveryTenPercent(DoubleConsumer progress, long
processedUnits, long totalUnits,
+ TenPercentCheckpoint checkpoint) {
+ reportEveryTenPercentInSlice(progress, 0.0, 1.0, processedUnits,
totalUnits, checkpoint);
+ }
+
+ static void reportEveryTenPercentInSlice(DoubleConsumer progress, double
sliceStart, double sliceEnd,
+ long processedUnits, long totalUnits, TenPercentCheckpoint checkpoint) {
+ if (progress == null) {
+ return;
+ }
+ long denominator = Math.max(1L, totalUnits);
+ double fraction = (double) processedUnits / denominator;
+ while (fraction >= checkpoint.nextCheckpoint) {
+ reportSlice(progress, sliceStart, sliceEnd, checkpoint.nextCheckpoint);
+ checkpoint.nextCheckpoint += TEN_PCT_STEP;
+ }
+ if (processedUnits >= totalUnits) {
+ reportSlice(progress, sliceStart, sliceEnd, 1.0);
+ }
Review Comment:
Once `processedUnits >= totalUnits`, this reports again on every following
row, and the `while` loop also keeps firing past 1.0. The totals are RocksDB
estimates (`getEstimatedKeyCount`, SST entry counts, clamped to 1), so the real
count can exceed them, and each report is a RocksDB write in `updateProgress`.
Every caller already reports the end of its slice after its loop, so the
trailing report can go:
```suggestion
while (checkpoint.nextCheckpoint <= 1.0 && fraction >=
checkpoint.nextCheckpoint) {
reportSlice(progress, sliceStart, sliceEnd, checkpoint.nextCheckpoint);
checkpoint.nextCheckpoint += TEN_PCT_STEP;
}
```
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java:
##########
@@ -1614,7 +1614,8 @@ private SubmitSnapshotDiffResponse submitSnapshotDiff(
submitSnapshotDiffRequest.getFromSnapshot(),
submitSnapshotDiffRequest.getToSnapshot(),
submitSnapshotDiffRequest.getForceFullDiff(),
- submitSnapshotDiffRequest.getDisableNativeDiff());
+ submitSnapshotDiffRequest.getDisableNativeDiff(),
+ submitSnapshotDiffRequest.hasOrdered() &&
submitSnapshotDiffRequest.getOrdered());
Review Comment:
nit: `getOrdered()` already returns false when the field is unset.
```suggestion
submitSnapshotDiffRequest.getOrdered());
```
--
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]
