This is an automated email from the ASF dual-hosted git repository.
jojochuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 7fe5962ce04 HDDS-16220. Reduce OM bucket write-lock hold time during
FSO directory purge (#11118)
7fe5962ce04 is described below
commit 7fe5962ce0474559a710ad87f823f9ca7d856516
Author: Andrey Yarovoy <[email protected]>
AuthorDate: Tue Sep 22 03:10:03 2026 -0400
HDDS-16220. Reduce OM bucket write-lock hold time during FSO directory
purge (#11118)
Generated-by: Claude Opus 4.8 <[email protected]>
---
.../apache/hadoop/hdds/utils/db/CodecBuffer.java | 9 +
.../TestOmMixedWorkloadUnderDeletionBench.java | 725 +++++++++++++++++++++
.../java/org/apache/hadoop/ozone/om/OMMetrics.java | 4 +-
.../key/OMDirectoriesPurgeRequestWithFSO.java | 564 +++++++++++-----
.../ozone/om/service/DirectoryDeletingService.java | 58 +-
.../key/TestOMDirectoriesPurgeApplyPerf.java | 226 +++++++
.../TestOMDirectoriesPurgeRequestAndResponse.java | 86 +++
.../om/service/TestDirectoryDeletingService.java | 60 +-
8 files changed, 1554 insertions(+), 178 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/db/CodecBuffer.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/db/CodecBuffer.java
index e9108112bd4..00de3bd2525 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/db/CodecBuffer.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/db/CodecBuffer.java
@@ -114,6 +114,15 @@ public static void enableLeakDetection() {
Factory.set(LeakDetector::newCodecBuffer, "LeakDetector::newCodecBuffer");
}
+ /**
+ * Restore the default (non-leak-detecting) buffer factory, reverting a
previous
+ * {@link #enableLeakDetection()}. Intended for performance-sensitive
measurements that need to avoid the
+ * per-allocation finalizer registration the leak detector adds.
+ */
+ public static void disableLeakDetection() {
+ Factory.set(CodecBuffer::new, "CodecBuffer::new");
+ }
+
/** The size of a buffer. */
public static class Capacity {
private final Object name;
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/service/TestOmMixedWorkloadUnderDeletionBench.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/service/TestOmMixedWorkloadUnderDeletionBench.java
new file mode 100644
index 00000000000..2f86b90eb99
--- /dev/null
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/service/TestOmMixedWorkloadUnderDeletionBench.java
@@ -0,0 +1,725 @@
+/*
+ * 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.service;
+
+import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
+import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL;
+import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_ITERATE_BATCH_SIZE;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.security.AccessController;
+import java.security.PrivilegedActionException;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+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 org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.StorageType;
+import org.apache.hadoop.hdds.utils.db.CodecBuffer;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.client.BucketArgs;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * End-to-end benchmark reproducing the interactive-workload degradation seen
when a large background deletion backlog
+ * is being reclaimed on the same bucket that clients are actively writing to.
+ *
+ * <p>Every OM write is applied on a single serial state-machine apply thread
under a per-bucket write lock. Reclaiming
+ * a deletion backlog drives {@code PurgeDirectories} transactions through
that thread; when the deleted directories are
+ * densely populated, a single purge batch moves a large number of
sub-files/sub-dirs under one write-lock hold, so the
+ * apply thread — and the bucket write lock — is occupied for the whole batch.
Concurrent user {@code create},
+ * {@code mkdir} and {@code rename} contend for that same lock and thread, and
reads contend for the bucket read lock,
+ * so their latency degrades while the backlog drains.
+ *
+ * <p>This benchmark stages two sets under a single volume and FSO bucket: a
densely-populated <em>backlog</em> subtree
+ * that is recursively deleted and drained, and a separate stable
<em>workload</em> set that is never deleted during the
+ * run (a pre-staged dataset the read ops resolve against, plus a per-thread
scratch area the write ops create into).
+ * Sharing the bucket is intentional — deletion and the client workload
contend on the same bucket lock, while the
+ * workload always hits live paths. It then compares a mixed client workload
on that bucket — create, mkdir, rename and
+ * a data-plane file write (create + write a block), plus a data-plane file
read and the metadata read RPCs that take
+ * the bucket read lock (getFileStatus, listStatus, getBucketInfo, lookupKey)
— in two conditions:
+ * <ul>
+ * <li><b>control</b> — no deletion running, and</li>
+ * <li><b>under load</b> — the same workload while the backlog subtree is
recursively deleted and fully purged from
+ * OM in the background (both FSO phases: moved into the deletedTable,
then purged back out),</li>
+ * </ul>
+ * reporting per-operation p50/p99 latency and the under-load degradation, so
the two code versions can be compared on
+ * how much apply-thread purge work bleeds into interactive latency. It also
reports the phase-1 drain time — the
+ * {@code DirectoryDeletingService} move into the deletedTable via {@code
OMDirectoriesPurgeRequestWithFSO}, the apply
+ * path this change optimizes — separately from the full both-phase drain, so
pure apply-thread throughput can be
+ * compared alongside the interactive degradation.
+ *
+ * <p>Deletion is configured with production-representative per-task limits so
batches are large. The {@code benchmark}
+ * tag is excluded from {@code mvn test} and CI by default, so it must be
re-enabled explicitly to run on demand
+ * (rebuild the reactor first to avoid stale-class errors):
+ * <pre>
+ * mvn -pl :ozone-integration-test test -DskipShade -DskipRecon \
+ * -Dtest=TestOmMixedWorkloadUnderDeletionBench -Dgroups=benchmark
-Dexcluded-test-groups= \
+ * -Dsurefire.failIfNoSpecifiedTests=false
-Djunit.jupiter.execution.timeout.default=20m
+ * </pre>
+ * <p>The timeout override belongs in the run command rather than a per-test
annotation: the repository-wide
+ * {@code junit.jupiter.execution.timeout.default} is 5 minutes, which is
shorter than staging the backlog plus the
+ * benchmark's own 900s phase-1 drain deadline, so without it a run is killed
mid-drain before that deadline reports
+ * anything.
+ *
+ * <p>Tunables: {@code bench.backlogDirs} (default 80), {@code
bench.backlogFilesPerDir} (default 1000),
+ * {@code bench.backlogNonEmptyEvery} (default 3 — every 3rd backlog file is
written with a block, the rest are
+ * empty so a large backlog stays cheap to stage), {@code bench.workloadDirs}
(default 20) and
+ * {@code bench.workloadFilesPerDir} (default 100) sizing the stable dataset
the read ops resolve against,
+ * {@code bench.fileBytes} (default 1 MiB) sizing the data-plane file
write/read payload (and the block-bearing
+ * staged files the reads pull), {@code bench.clientThreads} (default 4),
+ * {@code bench.opsPerThread} (default 400), {@code
bench.pathDeletingLimitPerTask} (default 2000) and
+ * {@code bench.keyDeletingLimitPerTask} (default 40000). The last two size
how much a single deletion round gathers;
+ * with the Ratis appender byte limit non-binding at these entry sizes, a
round's paths pack into one purge
+ * transaction, so raising them makes each apply move far more entries under a
single bucket write-lock hold — the
+ * regime where the apply-thread per-entry cost dominates interactive latency.
+ *
+ * <p>Adding {@code -Dbench.profile.event=<cpu|lock|wall|alloc>} profiles only
the under-load window with
+ * async-profiler, loaded reflectively from a local install whose paths must
be supplied via
+ * {@code -Dbench.profiler.jar} (the async-profiler jar) and {@code
-Dbench.profiler.lib} (its native library); the
+ * JFR is written under {@code -Dbench.profile.out}, default {@code /tmp}. For
accurate leaf frames also pass
+ * {@code -DargLine="-XX:+UnlockDiagnosticVMOptions -XX:+DebugNonSafepoints"}.
+ */
+@Tag("benchmark")
+public class TestOmMixedWorkloadUnderDeletionBench {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TestOmMixedWorkloadUnderDeletionBench.class);
+
+ private static final String OP_CREATE = "create";
+ private static final String OP_MKDIR = "mkdir";
+ private static final String OP_RENAME = "rename";
+ private static final String OP_FILEWRITE = "filewrite";
+ private static final String OP_FILEREAD = "fileread";
+ private static final String OP_GETFILESTATUS = "getfilestatus";
+ private static final String OP_LISTSTATUS = "liststatus";
+ private static final String OP_INFOBUCKET = "infobucket";
+ private static final String OP_GETKEYINFO = "getkeyinfo";
+ private static final String[] OPS =
+ {OP_CREATE, OP_MKDIR, OP_RENAME, OP_FILEWRITE, OP_FILEREAD,
+ OP_GETFILESTATUS, OP_LISTSTATUS, OP_INFOBUCKET, OP_GETKEYINFO};
+
+ // The three sandboxes share the parent /workload/bucket but are separate
subtrees; only the backlog is deleted.
+ // The deletion set (built, recursively deleted, then drained through the
apply thread).
+ private static final String BACKLOG_ROOT = "workload/bucket/backlog";
+ // The workload set — never deleted during the test: a stable pre-staged
dataset the read ops resolve against, plus
+ // a scratch area the write ops create into. Keeping this separate from the
deletion set means the concurrent
+ // operations always hit live paths while the backlog drains.
+ private static final String WORKLOAD_DATA_ROOT = "workload/bucket/data";
+ private static final String WORKLOAD_SCRATCH_ROOT =
"workload/bucket/scratch";
+
+ // Every bench.backlogNonEmptyEvery-th backlog file is written with a single
block so its KeyInfo carries a
+ // key-location list, exercising the block-metadata parse/serialize the
purge apply and flush paths hit in
+ // production. The rest are left empty because block-bearing files are far
more expensive to stage (block
+ // allocation + datanode write + commit), and a large backlog is what
actually stresses the apply thread.
+ private static final byte[] FILE_CONTENT = new byte[4];
+
+ /**
+ * Removes test-harness-only overhead that would otherwise distort the
apply/flush cost under measurement: the
+ * mini-cluster unconditionally enables {@link CodecBuffer} leak detection
(a per-allocation finalizer), and the test
+ * log config runs the {@code CodecBuffer}/managed-RocksDB loggers at
DEBUG/TRACE, which capture a full stack trace on
+ * every buffer allocation. Neither happens in a production OM running at
INFO.
+ *
+ * <p>Both settings are JVM-global, so the returned action must be run once
the benchmark is done to keep them from
+ * leaking into any later test sharing the same Maven fork.
+ */
+ private static Runnable stripTestOnlyOverhead() {
+ org.apache.log4j.Logger codecBufferLogger =
+
org.apache.log4j.Logger.getLogger("org.apache.hadoop.hdds.utils.db.CodecBuffer");
+ org.apache.log4j.Logger managedRocksLogger =
+
org.apache.log4j.Logger.getLogger("org.apache.hadoop.hdds.utils.db.managed");
+ org.apache.log4j.Level codecBufferLevel = codecBufferLogger.getLevel();
+ org.apache.log4j.Level managedRocksLevel = managedRocksLogger.getLevel();
+
+ CodecBuffer.disableLeakDetection();
+ codecBufferLogger.setLevel(org.apache.log4j.Level.INFO);
+ managedRocksLogger.setLevel(org.apache.log4j.Level.INFO);
+
+ return () -> {
+ CodecBuffer.enableLeakDetection();
+ codecBufferLogger.setLevel(codecBufferLevel);
+ managedRocksLogger.setLevel(managedRocksLevel);
+ };
+ }
+
+ @Test
+ public void benchmarkMixedWorkloadUnderDeletionLoad() throws Exception {
+ final String profileEvent = System.getProperty("bench.profile.event", "");
+
+ // Number of FSO buckets (in one volume) the backlog and workload are
spread across. With more than one bucket a
+ // background purge round gathers deleted dirs from several buckets, so an
ungrouped DirectoryDeletingService packs
+ // multiple buckets into one purge transaction and the apply path holds
all their write locks together; per-bucket
+ // grouping keeps each transaction single-bucket. backlogDirs below is the
TOTAL across buckets, split evenly.
+ final int numBuckets = Integer.getInteger("bench.numBuckets", 4);
+ final int backlogDirs = Integer.getInteger("bench.backlogDirs", 80);
+ final int backlogFilesPerDir =
Integer.getInteger("bench.backlogFilesPerDir", 1000);
+ final int nonEmptyEvery = Integer.getInteger("bench.backlogNonEmptyEvery",
3);
+ // Size of the stable workload dataset the read ops resolve against (never
deleted during the test).
+ final int workloadDirs = Integer.getInteger("bench.workloadDirs", 20);
+ final int workloadFilesPerDir =
Integer.getInteger("bench.workloadFilesPerDir", 100);
+ // Payload for the data-plane write/read ops and the block-bearing staged
workload files those reads hit.
+ final int fileBytes = Integer.getInteger("bench.fileBytes", 1024 * 1024);
+ final int clientThreads = Integer.getInteger("bench.clientThreads", 4);
+ final int opsPerThread = Integer.getInteger("bench.opsPerThread", 400);
+ final int pathDeletingLimit =
Integer.getInteger("bench.pathDeletingLimitPerTask", 2000);
+ final int keyDeletingLimit =
Integer.getInteger("bench.keyDeletingLimitPerTask", 40000);
+ // Phase-1 (DirectoryDeletingService) cadence. The interval is read in
MILLISECONDS (KeyManagerImpl), so 1000
+ // gives a genuine 1s gate between purge rounds: each round moves up to
pathDeletingLimit paths under one bucket
+ // write lock — the apply path this change optimizes — and gating spreads
phase 1 into a long, sampled window.
+ final int dirDeletingIntervalMs =
Integer.getInteger("bench.dirDeletingIntervalMs", 1000);
+ // Phase-2 (KeyDeletingService) is unchanged by this optimization; push
its interval past the phase-1 window so it
+ // does not run during measurement and only phase-1 contention is sampled.
+ final int blockDeletingIntervalSec =
Integer.getInteger("bench.blockDeletingIntervalSec", 600);
+ final String ratisAppenderByteLimit =
System.getProperty("bench.ratisAppenderByteLimit",
+ OMConfigKeys.OZONE_OM_RATIS_LOG_APPENDER_QUEUE_BYTE_LIMIT_DEFAULT);
+ final int perBucketBacklogDirs = Math.max(1, backlogDirs / numBuckets);
+ final int backlogFiles = numBuckets * perBucketBacklogDirs *
backlogFilesPerDir;
+
+ OzoneConfiguration conf = new OzoneConfiguration();
+ // Gate phase 1 (DirectoryDeletingService) at a genuine 1s interval so it
moves pathDeletingLimit paths per round
+ // under one apply-thread bucket write-lock hold — the path this change
optimizes — spreading phase 1 into a long,
+ // sampled window while the client workload runs. Phase 2
(KeyDeletingService) is unchanged by the optimization,
+ // so its interval is pushed past the window to keep it out of measurement.
+ conf.setInt(OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL,
dirDeletingIntervalMs);
+ conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL,
blockDeletingIntervalSec, TimeUnit.SECONDS);
+ conf.setInt(OMConfigKeys.OZONE_PATH_DELETING_LIMIT_PER_TASK,
pathDeletingLimit);
+ conf.setInt(OMConfigKeys.OZONE_KEY_DELETING_LIMIT_PER_TASK,
keyDeletingLimit);
+ conf.set(OMConfigKeys.OZONE_OM_RATIS_LOG_APPENDER_QUEUE_BYTE_LIMIT,
ratisAppenderByteLimit);
+ conf.setInt(OZONE_FS_ITERATE_BATCH_SIZE, 1000);
+
+ MiniOzoneCluster cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(3)
+ .build();
+ // Strip after build: MiniOzoneClusterImpl's static initializer enables
CodecBuffer leak detection when the class
+ // is first loaded here, so disabling it earlier would be undone before
any measurement.
+ Runnable restoreTestOnlyOverhead = stripTestOnlyOverhead();
+ try {
+ cluster.waitForClusterToBeReady();
+ DirectoryDeletingService dds =
cluster.getOzoneManager().getKeyManager().getDirDeletingService();
+ OMMetadataManager mm = cluster.getOzoneManager().getMetadataManager();
+
+ try (OzoneClient client = cluster.newClient()) {
+ BucketFs env = createBuckets(client, conf, numBuckets);
+ try {
+ // Stage the deletion set (recursively deleted and drained below)
and the stable workload set — a dataset the
+ // read ops resolve against plus a per-thread scratch area the write
ops create into — in every bucket, so the
+ // workload always hits live paths in the same buckets the purge is
draining. Backlog files carry only a tiny
+ // block (one key-location for the purge to parse); workload files
carry the full fileBytes payload.
+ byte[] fileContent = new byte[fileBytes];
+ for (int b = 0; b < numBuckets; b++) {
+ buildDenseTree(env.fs[b], new Path("/" + BACKLOG_ROOT),
perBucketBacklogDirs, backlogFilesPerDir,
+ nonEmptyEvery, FILE_CONTENT);
+ buildDenseTree(env.fs[b], new Path("/" + WORKLOAD_DATA_ROOT),
workloadDirs, workloadFilesPerDir,
+ nonEmptyEvery, fileContent);
+ env.fs[b].mkdirs(new Path("/" + WORKLOAD_SCRATCH_ROOT));
+ }
+ WorkloadDataset dataset = new WorkloadDataset(workloadDirs,
workloadFilesPerDir, nonEmptyEvery, fileContent);
+
+ // Control: interactive latency on the hot buckets with no deletion
running.
+ Percentiles[] control;
+ try (RunningWorkload controlWl = startMixedWorkload(env.fs,
env.volume, env.buckets, dataset, clientThreads,
+ opsPerThread, "control")) {
+ control = toPercentiles(controlWl.await(), "control");
+ }
+
+ // Profile only the under-load window when
-Dbench.profile.event=<cpu|lock|wall|alloc> is set. async-profiler
+ // is loaded reflectively from -Dbench.profiler.jar /
-Dbench.profiler.lib and a JFR recording is written to
+ // -Dbench.profile.out (default /tmp). For accurate leaf frames add
+ // -DargLine="-XX:+UnlockDiagnosticVMOptions
-XX:+DebugNonSafepoints".
+ Profiler profiler = profileEvent.isEmpty() ? null : Profiler.load();
+ String profileOut = null;
+ if (profiler != null) {
+ profileOut = System.getProperty("bench.profile.out", "/tmp") +
"/prof-mixed-" + profileEvent + ".jfr";
+ profiler.start(profileEvent, profileOut);
+ }
+
+ // Run the interactive workload continuously in background threads,
sampling throughout phase 1, and stop it
+ // the moment phase 1 completes so the under-load percentiles
reflect only the optimized, contended window.
+ // Phase 1 (DirectoryDeletingService moving sub-files/sub-dirs into
the deletedTable) takes the bucket write
+ // lock on the apply thread — the path this change optimizes; every
bucket's backlog is deleted here so purge
+ // rounds span buckets. Phase 2 (KeyDeletingService draining the
deletedTable) is unchanged by the change and
+ // is kept out of the window by a long block-deleting interval, so
it is neither sampled nor waited on.
+ Table<String, ?> deletedDirTable = mm.getDeletedDirTable();
+ long movedFilesBefore = dds.getMovedFilesCount();
+ long start = System.nanoTime();
+ long drainDeadline = start + TimeUnit.SECONDS.toNanos(900);
+ try (RunningWorkload underLoadWl = startMixedWorkload(env.fs,
env.volume, env.buckets, dataset,
+ clientThreads, 0, "under-load")) {
+ for (int b = 0; b < numBuckets; b++) {
+ env.fs[b].delete(new Path("/" + BACKLOG_ROOT), true);
+ }
+ LOG.info("delete(recursive) issued on {} buckets; backlog draining
in background", numBuckets);
+
+ long phase1DrainNanos;
+ try {
+ phase1DrainNanos = awaitPhase1Drain(dds, mm, deletedDirTable,
movedFilesBefore, backlogFiles,
+ underLoadWl, start, drainDeadline);
+ } finally {
+ underLoadWl.stop();
+ if (profiler != null) {
+ profiler.stop();
+ }
+ }
+ double phase1DrainMs = phase1DrainNanos / 1_000_000.0;
+ List<List<Long>> underLoadSamples = underLoadWl.await();
+ Percentiles[] underLoad = toPercentiles(underLoadSamples,
"under-load");
+ long underLoadOps = totalOps(underLoadSamples);
+ String header = String.format(Locale.ROOT,
+ "BENCH mixed numBuckets=%d backlogFiles=%d threads=%d
opsPerThread=%d ratisByteLimit=%s "
+ + "underLoadOps=%d phase1DrainMs=%.1f",
+ numBuckets, backlogFiles, clientThreads, opsPerThread,
ratisAppenderByteLimit, underLoadOps,
+ phase1DrainMs);
+ printBenchLine(control, underLoad, header);
+ if (profileOut != null) {
+ System.out.printf(Locale.ROOT, "BENCH profile event=%s
out=%s%n", profileEvent, profileOut);
+ }
+ }
+ } finally {
+ for (FileSystem f : env.fs) {
+ org.apache.hadoop.io.IOUtils.closeStream(f);
+ }
+ }
+ }
+ } finally {
+ // Restore last, so the buffer-heavy cluster shutdown still runs without
the leak detector's per-allocation
+ // finalizer, and restore unconditionally so a failing shutdown cannot
leak the stripped state into later tests.
+ try {
+ cluster.shutdown();
+ } finally {
+ restoreTestOnlyOverhead.run();
+ }
+ }
+ }
+
+ /**
+ * Blocks until phase 1 (DirectoryDeletingService) finishes: every backlog
sub-file has been moved (cheap counter)
+ * and the deletedDirTable has been fully drained of sub-dirs.
countRowsInTable scans the table, so it is only probed
+ * once the moved-files counter shows the sub-files are all moved. Phase 2
(deletedTable purge) is out of scope.
+ * Returns the drain duration in nanos, or -1 if the workload failed before
the drain completed.
+ */
+ @SuppressWarnings("checkstyle:ParameterNumber")
+ private static long awaitPhase1Drain(DirectoryDeletingService dds,
OMMetadataManager mm,
+ Table<String, ?> deletedDirTable, long movedFilesBefore, long
backlogFiles, RunningWorkload underLoadWl,
+ long start, long drainDeadline) throws Exception {
+ while (true) {
+ long moved = dds.getMovedFilesCount() - movedFilesBefore;
+ if (moved >= backlogFiles && mm.countRowsInTable(deletedDirTable) == 0) {
+ return System.nanoTime() - start;
+ }
+ if (underLoadWl.failed()) {
+ return -1;
+ }
+ if (System.nanoTime() > drainDeadline) {
+ throw new IllegalStateException("phase 1 did not drain within 900s:
moved=" + moved
+ + " expected=" + backlogFiles
+ + " deletedDirTableRows=" + mm.countRowsInTable(deletedDirTable));
+ }
+ Thread.sleep(200);
+ }
+ }
+
+ private void buildDenseTree(FileSystem fs, Path root, int dirs, int
filesPerDir, int nonEmptyEvery,
+ byte[] blockContent) throws Exception {
+ long buildStart = System.nanoTime();
+ // Staging is client/RPC-round-trip bound, not apply-thread bound, so more
concurrent creators speed it up
+ // nearly linearly; a large backlog is otherwise the long pole of the run.
Setup-only — does not affect the
+ // measured control/under-load workload.
+ ExecutorService pool =
Executors.newFixedThreadPool(Integer.getInteger("bench.stagingThreads", 48));
+ List<Future<?>> futures = new ArrayList<>(dirs);
+ for (int d = 0; d < dirs; d++) {
+ final Path dir = new Path(root, "dir" + d);
+ futures.add(pool.submit(() -> {
+ fs.mkdirs(dir);
+ for (int f = 0; f < filesPerDir; f++) {
+ Path file = new Path(dir, "f" + f);
+ if (nonEmptyEvery > 0 && f % nonEmptyEvery == 0) {
+ try (FSDataOutputStream os = fs.create(file, true)) {
+ os.write(blockContent);
+ }
+ } else {
+ fs.create(file, true).close();
+ }
+ }
+ return null;
+ }));
+ }
+ try {
+ for (Future<?> future : futures) {
+ future.get();
+ }
+ } finally {
+ HadoopExecutors.shutdown(pool, LOG, 120, TimeUnit.SECONDS);
+ }
+ LOG.info("Built dense backlog: {} files in {} dirs, every {}-th with a
block ({} ms)",
+ dirs * filesPerDir, dirs, nonEmptyEvery, (System.nanoTime() -
buildStart) / 1_000_000);
+ }
+
+ /** Creates one volume with {@code numBuckets} FSO buckets and one {@code
o3fs} FileSystem rooted at each bucket, so
+ * the backlog and workload can be spread across buckets and a purge round
can gather deleted dirs from several of
+ * them. Returned FileSystems must be closed by the caller. */
+ private BucketFs createBuckets(OzoneClient client, OzoneConfiguration conf,
int numBuckets) throws IOException {
+ OzoneBucket first =
org.apache.hadoop.ozone.DataTestUtil.createVolumeAndBucket(client,
+ BucketLayout.FILE_SYSTEM_OPTIMIZED);
+ OzoneVolume volume =
client.getObjectStore().getVolume(first.getVolumeName());
+ OzoneBucket[] buckets = new OzoneBucket[numBuckets];
+ FileSystem[] fs = new FileSystem[numBuckets];
+ buckets[0] = first;
+ BucketArgs args = BucketArgs.newBuilder().setStorageType(StorageType.DISK)
+ .setBucketLayout(BucketLayout.FILE_SYSTEM_OPTIMIZED).build();
+ for (int b = 1; b < numBuckets; b++) {
+ String name = first.getName() + "-" + b;
+ volume.createBucket(name, args);
+ buckets[b] = volume.getBucket(name);
+ }
+ for (int b = 0; b < numBuckets; b++) {
+ OzoneConfiguration bucketConf = new OzoneConfiguration(conf);
+ bucketConf.set(FS_DEFAULT_NAME_KEY, String.format("%s://%s.%s/",
OzoneConsts.OZONE_URI_SCHEME,
+ buckets[b].getName(), volume.getName()));
+ fs[b] = FileSystem.get(bucketConf);
+ }
+ return new BucketFs(volume, buckets, fs);
+ }
+
+ /** One volume, its FSO buckets, and the per-bucket {@code o3fs} FileSystems
(index-aligned with {@code buckets}). */
+ private static final class BucketFs {
+ private final OzoneVolume volume;
+ private final OzoneBucket[] buckets;
+ private final FileSystem[] fs;
+
+ BucketFs(OzoneVolume volume, OzoneBucket[] buckets, FileSystem[] fs) {
+ this.volume = volume;
+ this.buckets = buckets;
+ this.fs = fs;
+ }
+ }
+
+ /**
+ * Starts {@code threads} client threads spread round-robin across the hot
buckets (thread {@code t} is pinned to
+ * bucket {@code t % buckets.length} and its FileSystem), each cycling
through the {@link #OPS} mix (create, mkdir,
+ * rename and a data-plane file write, plus a data-plane file read and the
metadata read RPCs that contend for the
+ * bucket read lock: getFileStatus, listStatus, getBucketInfo and lookupKey)
and recording per-operation nanosecond
+ * samples. When {@code opsPerThread > 0} each thread runs that fixed number
of ops (control pass); when
+ * {@code opsPerThread <= 0} each thread runs continuously until {@link
RunningWorkload#stop()} is called (the
+ * under-load drain window). Call {@link RunningWorkload#await()} to join
the threads and collect the samples
+ * grouped by operation index (aligned with {@link #OPS}).
+ */
+ private RunningWorkload startMixedWorkload(FileSystem[] fs, OzoneVolume
volume, OzoneBucket[] buckets,
+ WorkloadDataset dataset, int threads, int opsPerThread, String label) {
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ AtomicBoolean failed = new AtomicBoolean(false);
+ AtomicBoolean running = new AtomicBoolean(true);
+ List<Future<List<long[]>>> futures = new ArrayList<>(threads);
+ for (int t = 0; t < threads; t++) {
+ final int threadId = t;
+ final int bucketId = t % buckets.length;
+ final FileSystem tfs = fs[bucketId];
+ final OzoneBucket bucket = buckets[bucketId];
+ final String bucketName = bucket.getName();
+ futures.add(pool.submit((Callable<List<long[]>>) () -> {
+ List<long[]> latencies = new ArrayList<>(Math.max(16, opsPerThread));
+ // Write ops create into this thread's private scratch subtree; read
ops resolve against the pre-staged
+ // dataset. Both live under the workload set (same volume and bucket
as the backlog), which is never deleted.
+ Path scratch = new Path("/" + WORKLOAD_SCRATCH_ROOT, label + "-" +
threadId);
+ tfs.mkdirs(scratch);
+ Path lastCreated = null;
+ byte[] readBuf = new byte[64 * 1024];
+ startLatch.await();
+ // Fixed batch when opsPerThread > 0 (control); otherwise loop until
stop() clears running (under-load).
+ for (int i = 0; running.get() && (opsPerThread <= 0 || i <
opsPerThread); i++) {
+ int opIdx = i % OPS.length;
+ // rename operates on the file created in this thread's previous
create op; fall back to a create when none
+ // is pending so the cycle never renames a missing path.
+ if (opIdx == 2 && lastCreated == null) {
+ opIdx = 0;
+ }
+ // Read ops target an existing dataset entry; the index spread keeps
threads off a single dir.
+ int dataDir = (threadId + i) % dataset.dirs;
+ int dataFile = i % dataset.filesPerDir;
+ // The data-plane read must hit a staged file that carries a block
(every nonEmptyEvery-th file).
+ int blockFile = (i % Math.max(1, dataset.filesPerDir /
dataset.nonEmptyEvery)) * dataset.nonEmptyEvery;
+ long t0 = System.nanoTime();
+ try {
+ switch (opIdx) {
+ case 0:
+ Path created = new Path(scratch, "f" + i);
+ tfs.create(created, true).close();
+ lastCreated = created;
+ break;
+ case 1:
+ tfs.mkdirs(new Path(scratch, "d" + i));
+ break;
+ case 2:
+ tfs.rename(lastCreated, new Path(scratch, "f" + i + "-r"));
+ lastCreated = null;
+ break;
+ case 3:
+ // File write: create + write a real block, exercising the
data-plane write path (block allocation,
+ // datanode write, commit) on top of the OM key create/commit.
+ try (FSDataOutputStream os = tfs.create(new Path(scratch, "w" +
i), true)) {
+ os.write(dataset.fileContent);
+ }
+ break;
+ case 4:
+ // File read: open a staged block-bearing file and read it
fully, exercising the data-plane read path
+ // (block-location lookup + datanode read).
+ readFully(tfs, new Path("/" + WORKLOAD_DATA_ROOT + "/dir" +
dataDir, "f" + blockFile), readBuf);
+ break;
+ case 5:
+ tfs.getFileStatus(new Path("/" + WORKLOAD_DATA_ROOT + "/dir" +
dataDir, "f" + dataFile));
+ break;
+ case 6:
+ tfs.listStatus(new Path("/" + WORKLOAD_DATA_ROOT, "dir" +
dataDir));
+ break;
+ case 7:
+ volume.getBucket(bucketName);
+ break;
+ default:
+ bucket.getKey(WORKLOAD_DATA_ROOT + "/dir" + dataDir + "/f" +
dataFile);
+ break;
+ }
+ } catch (IOException | RuntimeException e) {
+ failed.set(true);
+ throw e;
+ }
+ latencies.add(new long[] {opIdx, System.nanoTime() - t0});
+ }
+ return latencies;
+ }));
+ }
+ startLatch.countDown();
+ return new RunningWorkload(pool, futures, running, failed, label);
+ }
+
+ /**
+ * Handle to a running {@link #startMixedWorkload} run: stop the threads,
then await and collect their samples.
+ * Closing stops the threads and shuts the pool down, so a failure anywhere
in the measured window cannot leave the
+ * non-daemon client threads running and keep the Maven fork alive.
+ */
+ private static final class RunningWorkload implements AutoCloseable {
+ private final ExecutorService pool;
+ private final List<Future<List<long[]>>> futures;
+ private final AtomicBoolean running;
+ private final AtomicBoolean failed;
+ private final String label;
+
+ RunningWorkload(ExecutorService pool, List<Future<List<long[]>>> futures,
AtomicBoolean running,
+ AtomicBoolean failed, String label) {
+ this.pool = pool;
+ this.futures = futures;
+ this.running = running;
+ this.failed = failed;
+ this.label = label;
+ }
+
+ void stop() {
+ running.set(false);
+ }
+
+ boolean failed() {
+ return failed.get();
+ }
+
+ @Override
+ public void close() {
+ stop();
+ HadoopExecutors.shutdown(pool, LOG, 180, TimeUnit.SECONDS);
+ }
+
+ /** Joins the client threads and returns per-operation nanosecond samples
grouped by operation index. */
+ List<List<Long>> await() throws Exception {
+ List<List<Long>> byOp = new ArrayList<>(OPS.length);
+ for (String ignored : OPS) {
+ byOp.add(new ArrayList<>());
+ }
+ for (Future<List<long[]>> future : futures) {
+ for (long[] sample : future.get()) {
+ byOp.get((int) sample[0]).add(sample[1]);
+ }
+ }
+ if (failed.get()) {
+ throw new IllegalStateException("client thread failed during " +
label);
+ }
+ for (int op = 0; op < OPS.length; op++) {
+ LOG.info("{} {}: ops={}", label, OPS[op], byOp.get(op).size());
+ }
+ return byOp;
+ }
+ }
+
+ /** Total number of samples across all operations (how many ops backed the
reported percentiles). */
+ private static long totalOps(List<List<Long>> byOp) {
+ long total = 0;
+ for (List<Long> opSamples : byOp) {
+ total += opSamples.size();
+ }
+ return total;
+ }
+
+ /** Reduces per-operation nanosecond samples to {@link Percentiles}
(index-aligned with {@link #OPS}). */
+ private static Percentiles[] toPercentiles(List<List<Long>> byOp, String
label) {
+ Percentiles[] result = new Percentiles[OPS.length];
+ for (int op = 0; op < OPS.length; op++) {
+ result[op] = Percentiles.of(byOp.get(op));
+ LOG.info("{} {}: ops={} p50Ms={} p99Ms={}", label, OPS[op],
byOp.get(op).size(),
+ result[op].p50, result[op].p99);
+ }
+ return result;
+ }
+
+ private static double safeRatio(double num, double den) {
+ return den <= 0 ? Double.NaN : num / den;
+ }
+
+ /** Appends the per-operation control/under-load percentiles and degradation
ratios to {@code header} and prints
+ * the single machine-readable {@code BENCH mixed ...} line the A/B driver
greps for. */
+ private static void printBenchLine(Percentiles[] control, Percentiles[]
underLoad, String header) {
+ StringBuilder line = new StringBuilder(header);
+ for (int op = 0; op < OPS.length; op++) {
+ line.append(String.format(Locale.ROOT, " %s[control_p50=%.2f
control_p99=%.2f "
+ + "underLoad_p50=%.2f underLoad_p99=%.2f deg50=%.2fx
deg99=%.2fx]",
+ OPS[op], control[op].p50, control[op].p99, underLoad[op].p50,
underLoad[op].p99,
+ safeRatio(underLoad[op].p50, control[op].p50),
safeRatio(underLoad[op].p99, control[op].p99)));
+ }
+ System.out.println(line);
+ }
+
+ /** Reads {@code p} to end into the reusable {@code buf}, discarding the
bytes — the benchmark measures the read
+ * path (block-location lookup + datanode read), not the payload. */
+ private static void readFully(FileSystem fs, Path p, byte[] buf) throws
IOException {
+ try (FSDataInputStream in = fs.open(p)) {
+ int n;
+ do {
+ n = in.read(buf);
+ } while (n != -1);
+ }
+ }
+
+ /** Shape of the stable workload dataset the read ops resolve against
({@code dirs} directories, each holding
+ * {@code filesPerDir} files under {@link #WORKLOAD_DATA_ROOT}, every {@code
nonEmptyEvery}-th carrying a
+ * {@code fileContent} block), plus the payload the data-plane file write op
writes. */
+ private static final class WorkloadDataset {
+ private final int dirs;
+ private final int filesPerDir;
+ private final int nonEmptyEvery;
+ private final byte[] fileContent;
+
+ WorkloadDataset(int dirs, int filesPerDir, int nonEmptyEvery, byte[]
fileContent) {
+ this.dirs = dirs;
+ this.filesPerDir = filesPerDir;
+ this.nonEmptyEvery = nonEmptyEvery;
+ this.fileContent = fileContent;
+ }
+ }
+
+ /**
+ * Thin reflective wrapper over async-profiler's {@code
one.profiler.AsyncProfiler} API, loaded at runtime from the
+ * configured jar so the benchmark carries no compile-time or Maven
dependency on async-profiler.
+ */
+ private static final class Profiler {
+ private final Object delegate;
+ private final Method execute;
+
+ private Profiler(Object delegate, Method execute) {
+ this.delegate = delegate;
+ this.execute = execute;
+ }
+
+ static Profiler load() throws Exception {
+ String jar = requireProfilerPath("bench.profiler.jar");
+ String lib = requireProfilerPath("bench.profiler.lib");
+ try {
+ return
AccessController.doPrivileged((PrivilegedExceptionAction<Profiler>) () -> {
+ URLClassLoader loader = new URLClassLoader(new URL[] {new
File(jar).toURI().toURL()},
+ Profiler.class.getClassLoader());
+ Class<?> clazz = Class.forName("one.profiler.AsyncProfiler", true,
loader);
+ Object instance = clazz.getMethod("getInstance",
String.class).invoke(null, lib);
+ return new Profiler(instance, clazz.getMethod("execute",
String.class));
+ });
+ } catch (PrivilegedActionException e) {
+ throw (Exception) e.getCause();
+ }
+ }
+
+ private static String requireProfilerPath(String property) {
+ String value = System.getProperty(property);
+ if (value == null || value.isEmpty()) {
+ throw new IllegalStateException("bench.profile.event is set but " +
property + " is not; point it at your "
+ + "local async-profiler install (jar and native library) to enable
profiling");
+ }
+ return value;
+ }
+
+ void start(String event, String jfrFile) throws Exception {
+ // JFR output (inferred from the .jfr extension) records cpu/wall sample
intervals and every contended lock/alloc
+ // event, so a single recording converts to both a flamegraph and
collapsed stacks afterwards.
+ execute.invoke(delegate, String.format(Locale.ROOT,
"start,event=%s,interval=1ms,file=%s", event, jfrFile));
+ }
+
+ void stop() throws Exception {
+ execute.invoke(delegate, "stop");
+ }
+ }
+
+ /** p50/p99 in milliseconds over a set of nanosecond latency samples. */
+ private static final class Percentiles {
+ private final double p50;
+ private final double p99;
+
+ private Percentiles(double p50, double p99) {
+ this.p50 = p50;
+ this.p99 = p99;
+ }
+
+ static Percentiles of(List<Long> samplesNs) {
+ if (samplesNs.isEmpty()) {
+ return new Percentiles(Double.NaN, Double.NaN);
+ }
+ long[] sorted =
samplesNs.stream().mapToLong(Long::longValue).sorted().toArray();
+ double p50 = sorted[(int) (sorted.length * 0.50)] / 1_000_000.0;
+ double p99 = sorted[Math.min(sorted.length - 1, (int) (sorted.length *
0.99))] / 1_000_000.0;
+ return new Percentiles(p50, p99);
+ }
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMMetrics.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMMetrics.java
index 426501c1866..57697fb4b01 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMMetrics.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMMetrics.java
@@ -883,8 +883,8 @@ public void incNumKeyDeletes() {
numKeyDeletes.incr();
}
- public void incNumKeyDeletesInternal() {
- numKeyDeletes.incr();
+ public void incNumKeyDeletesInternal(long count) {
+ numKeyDeletes.incr(count);
}
public void incNumKeyDeletes(int count) {
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMDirectoriesPurgeRequestWithFSO.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMDirectoriesPurgeRequestWithFSO.java
index e2ca3a76136..34a47f1c256 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMDirectoriesPurgeRequestWithFSO.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMDirectoriesPurgeRequestWithFSO.java
@@ -23,6 +23,7 @@
import static
org.apache.hadoop.ozone.om.snapshot.SnapshotUtils.validatePreviousSnapshotId;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
@@ -34,6 +35,8 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
@@ -42,7 +45,6 @@
import org.apache.hadoop.ozone.audit.AuditLoggerType;
import org.apache.hadoop.ozone.audit.OMSystemAction;
import org.apache.hadoop.ozone.om.DeletingServiceMetrics;
-import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.OMMetadataManager.VolumeBucketId;
import org.apache.hadoop.ozone.om.OMMetrics;
import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
@@ -52,6 +54,8 @@
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils;
+import org.apache.hadoop.ozone.om.helpers.QuotaUtil;
import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
import org.apache.hadoop.ozone.om.response.OMClientResponse;
@@ -81,7 +85,6 @@ public OMDirectoriesPurgeRequestWithFSO(OMRequest omRequest) {
}
@Override
- @SuppressWarnings("methodlength")
public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
ExecutionContext context) {
PurgeDirectoriesRequest purgeDirsRequest =
getOmRequest().getPurgeDirectoriesRequest();
@@ -90,41 +93,38 @@ public OMClientResponse validateAndUpdateCache(OzoneManager
ozoneManager, Execut
List<OzoneManagerProtocolProtos.PurgePathRequest> purgeRequests =
purgeDirsRequest.getDeletedPathList();
- Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap = new HashMap<>();
OmMetadataManagerImpl omMetadataManager = (OmMetadataManagerImpl)
ozoneManager.getMetadataManager();
- Map<String, OmKeyInfo> openKeyInfoMap = new HashMap<>();
OMMetrics omMetrics = ozoneManager.getMetrics();
DeletingServiceMetrics deletingServiceMetrics =
ozoneManager.getDeletionMetrics();
OMResponse.Builder omResponse = OmResponseUtil.getOMResponseBuilder(
getOmRequest());
final SnapshotInfo fromSnapshotInfo;
- Set<String> subDirNames = new HashSet<>();
- Set<String> subFileNames = new HashSet<>();
- Set<String> deletedDirNames = new HashSet<>();
-
try {
- fromSnapshotInfo = fromSnapshot != null ?
SnapshotUtils.getSnapshotInfo(ozoneManager,
- fromSnapshot) : null;
- // Checking if this request is an old request or new one.
- if (purgeDirsRequest.hasExpectedPreviousSnapshotID()) {
- // Validating previous snapshot since while purging deletes, a
snapshot create request could make this purge
- // directory request invalid on AOS since the deletedDirectory would
be in the newly created snapshot. Adding
- // subdirectories could lead to not being able to reclaim sub-files
and subdirectories since the
- // file/directory would be present in the newly created snapshot.
- // Validating previous snapshot can ensure the chain hasn't changed.
- UUID expectedPreviousSnapshotId =
purgeDirsRequest.getExpectedPreviousSnapshotID().hasUuid()
- ?
fromProtobuf(purgeDirsRequest.getExpectedPreviousSnapshotID().getUuid()) : null;
- validatePreviousSnapshotId(fromSnapshotInfo,
omMetadataManager.getSnapshotChainManager(),
- expectedPreviousSnapshotId);
- }
+ fromSnapshotInfo = resolveFromSnapshotInfo(ozoneManager,
omMetadataManager, purgeDirsRequest, fromSnapshot);
} catch (IOException e) {
- LOG.error("Error occurred while performing OMDirectoriesPurge. ", e);
- if (LOG.isDebugEnabled()) {
-
AUDIT.logWriteFailure(ozoneManager.buildAuditMessageForFailure(OMSystemAction.DIRECTORY_DELETION,
null, e));
+ return directoryPurgeFailure(ozoneManager, omResponse, e);
+ }
+ // Phase 1 (no lock): parse every purge entry and precompute its delete
key, path key, replicated size and any
+ // hsync open-key name. None of this depends on the bucket write lock, so
doing it up front keeps the string and
+ // protobuf work out of the critical section and shortens the write-lock
hold.
+ Map<VolumeBucketId, BucketNameInfo> volumeBucketIdMap =
purgeDirsRequest.getBucketNameInfosList().stream()
+ .collect(Collectors.toMap(bucketNameInfo ->
+ new VolumeBucketId(bucketNameInfo.getVolumeId(),
bucketNameInfo.getBucketId()),
+ Function.identity()));
+ List<PreparedEntry> preparedSubDirs = new ArrayList<>();
+ List<PreparedEntry> preparedSubFiles = new ArrayList<>();
+ List<PreparedDirPurge> preparedDirPurges = new ArrayList<>();
+ for (OzoneManagerProtocolProtos.PurgePathRequest path : purgeRequests) {
+ preparedSubDirs.addAll(prepareMarkDeletedSubDirs(path,
omMetadataManager));
+ preparedSubFiles.addAll(prepareMoveDeletedSubFiles(path,
omMetadataManager));
+ if (path.hasDeletedDir()) {
+ BucketNameInfo bucketNameInfo = volumeBucketIdMap.get(new
VolumeBucketId(path.getVolumeId(),
+ path.getBucketId()));
+ preparedDirPurges.add(prepareDeletedDir(path, bucketNameInfo));
}
- return new
OMDirectoriesPurgeResponseWithFSO(createErrorOMResponse(omResponse, e));
}
+
List<String[]> bucketLockKeys = getBucketLockKeySet(purgeDirsRequest);
mergeOmLockDetails(omMetadataManager.getLock().acquireWriteLocks(BUCKET_LOCK,
bucketLockKeys));
boolean lockAcquired = getOmLockDetails().isLockAcquired();
@@ -135,102 +135,27 @@ public OMClientResponse
validateAndUpdateCache(OzoneManager ozoneManager, Execut
AUDIT.logWriteFailure(ozoneManager.buildAuditMessageForFailure(OMSystemAction.DIRECTORY_DELETION,
null, oe));
return new
OMDirectoriesPurgeResponseWithFSO(createErrorOMResponse(omResponse, oe));
}
+ PurgeApplyResult result = new PurgeApplyResult();
try {
- int numSubDirMoved = 0, numSubFilesMoved = 0, numDirsDeleted = 0;
- Map<VolumeBucketId, BucketNameInfo> volumeBucketIdMap =
purgeDirsRequest.getBucketNameInfosList().stream()
- .collect(Collectors.toMap(bucketNameInfo ->
- new VolumeBucketId(bucketNameInfo.getVolumeId(),
bucketNameInfo.getBucketId()),
- Function.identity()));
- for (OzoneManagerProtocolProtos.PurgePathRequest path : purgeRequests) {
- for (OzoneManagerProtocolProtos.KeyInfo key :
path.getMarkDeletedSubDirsList()) {
- ProcessedKeyInfo processed = processDeleteKey(key, path,
omMetadataManager);
- subDirNames.add(processed.deleteKey);
-
- omMetrics.decNumKeys();
- omMetrics.incNumKeyDeletesInternal();
- OmBucketInfo omBucketInfo = accumulatedBucketInfo(volBucketInfoMap,
omMetadataManager,
- processed.volumeName, processed.bucketName);
- // bucketInfo can be null in case of delete volume or bucket
- // or key does not belong to bucket as bucket is recreated
- if (null != omBucketInfo && omBucketInfo.getObjectID() ==
path.getBucketId()) {
- omBucketInfo.decrUsedNamespace(1L, true);
- String ozoneDbKey =
omMetadataManager.getOzonePathKey(path.getVolumeId(),
- path.getBucketId(), processed.keyInfo.getParentObjectID(),
- processed.keyInfo.getFileName());
- omMetadataManager.getDirectoryTable().addCacheEntry(new
CacheKey<>(ozoneDbKey),
- CacheValue.get(context.getIndex()));
- volBucketInfoMap.putIfAbsent(processed.volBucketPair,
omBucketInfo);
- }
- }
-
- for (OzoneManagerProtocolProtos.KeyInfo key :
path.getDeletedSubFilesList()) {
- ProcessedKeyInfo processed = processDeleteKey(key, path,
omMetadataManager);
- subFileNames.add(processed.deleteKey);
-
- // If omKeyInfo has hsync metadata, delete its corresponding open
key as well
- String dbOpenKey;
- String hsyncClientId =
processed.keyInfo.getMetadata().get(OzoneConsts.HSYNC_CLIENT_ID);
- if (hsyncClientId != null) {
- long parentId = processed.keyInfo.getParentObjectID();
- dbOpenKey = omMetadataManager.getOpenFileName(path.getVolumeId(),
path.getBucketId(),
- parentId, processed.keyInfo.getFileName(), hsyncClientId);
- OmKeyInfo openKeyInfo =
omMetadataManager.getOpenKeyTable(getBucketLayout()).get(dbOpenKey);
- if (openKeyInfo != null) {
- openKeyInfo = openKeyInfo.withMetadataMutations(
- metadata -> metadata.put(DELETED_HSYNC_KEY, "true"));
- openKeyInfoMap.put(dbOpenKey, openKeyInfo);
- }
- }
-
- omMetrics.decNumKeys();
- omMetrics.incNumKeyDeletesInternal();
- numSubFilesMoved++;
- OmBucketInfo omBucketInfo = accumulatedBucketInfo(volBucketInfoMap,
omMetadataManager,
- processed.volumeName, processed.bucketName);
- // bucketInfo can be null in case of delete volume or bucket
- // or key does not belong to bucket as bucket is recreated
- if (null != omBucketInfo
- && omBucketInfo.getObjectID() == path.getBucketId()) {
- long totalSize = sumBlockLengths(processed.keyInfo);
- omBucketInfo.decrUsedBytes(totalSize, true);
- omBucketInfo.decrUsedNamespace(1L, true);
- String ozoneDbKey =
omMetadataManager.getOzonePathKey(path.getVolumeId(),
- path.getBucketId(), processed.keyInfo.getParentObjectID(),
- processed.keyInfo.getFileName());
- omMetadataManager.getFileTable().addCacheEntry(new
CacheKey<>(ozoneDbKey),
- CacheValue.get(context.getIndex()));
- volBucketInfoMap.putIfAbsent(processed.volBucketPair,
omBucketInfo);
- }
- }
- if (path.hasDeletedDir()) {
- deletedDirNames.add(path.getDeletedDir());
- BucketNameInfo bucketNameInfo = volumeBucketIdMap.get(new
VolumeBucketId(path.getVolumeId(),
- path.getBucketId()));
- OmBucketInfo omBucketInfo = accumulatedBucketInfo(volBucketInfoMap,
omMetadataManager,
- bucketNameInfo.getVolumeName(), bucketNameInfo.getBucketName());
- if (omBucketInfo != null && omBucketInfo.getObjectID() ==
path.getBucketId()) {
- omBucketInfo.purgeSnapshotUsedNamespace(1);
- volBucketInfoMap.put(Pair.of(omBucketInfo.getVolumeName(),
omBucketInfo.getBucketName()), omBucketInfo);
- }
- numDirsDeleted++;
- }
- }
+ // Phase 2 (under the bucket write lock): apply the prepared cache
tombstones, hsync open-key cleanup and
+ // aggregated per-bucket quota changes.
+ applyPreparedEntries(preparedSubDirs, preparedSubFiles,
preparedDirPurges, omMetadataManager,
+ context.getIndex(), result);
- // Remove deletedDirNames from subDirNames to avoid duplication
- subDirNames.removeAll(deletedDirNames);
- numSubDirMoved = subDirNames.size();
- deletingServiceMetrics.incrNumSubDirectoriesMoved(numSubDirMoved);
- deletingServiceMetrics.incrNumSubFilesMoved(numSubFilesMoved);
- deletingServiceMetrics.incrNumDirPurged(numDirsDeleted);
-
- // All bucket locks are still held here, and every purge path has been
processed, so the
- // accumulated copies can be published as a whole.
- for (Map.Entry<Pair<String, String>, OmBucketInfo> entry :
volBucketInfoMap.entrySet()) {
+ // All bucket locks are still held here, and every purge path has been
processed, so the accumulated
+ // copies can be published as a whole. Publishing only here, after all
fallible apply work has
+ // succeeded, keeps a failed purge from changing the live bucket usage
(HDDS-16161).
+ for (Map.Entry<Pair<String, String>, OmBucketInfo> entry :
result.volBucketInfoMap.entrySet()) {
omMetadataManager.getBucketTable().addCacheEntry(
omMetadataManager.getBucketKey(entry.getKey().getLeft(),
entry.getKey().getRight()),
entry.getValue(), context.getIndex());
}
+ // The per-entry global OM metric mutations were unconditional, so the
count is the total number of prepared
+ // sub-directories and sub-files.
+ long numKeysProcessed = (long) preparedSubDirs.size() +
preparedSubFiles.size();
+ int numSubDirMoved = recordDeletionMetrics(deletingServiceMetrics,
omMetrics, numKeysProcessed, result);
+
TransactionInfo transactionInfo =
TransactionInfo.valueOf(context.getTermIndex());
if (fromSnapshotInfo != null) {
fromSnapshotInfo.setLastTransactionInfo(transactionInfo.toByteString());
@@ -243,17 +168,7 @@ public OMClientResponse
validateAndUpdateCache(OzoneManager ozoneManager, Execut
}
if (LOG.isDebugEnabled()) {
- Map<String, String> auditParams = new LinkedHashMap<>();
- if (fromSnapshotInfo != null) {
- auditParams.put(AUDIT_PARAM_SNAPSHOT_ID,
fromSnapshotInfo.getSnapshotId().toString());
- }
- auditParams.put(AUDIT_PARAM_DIRS_DELETED,
String.valueOf(numDirsDeleted));
- auditParams.put(AUDIT_PARAM_SUBDIRS_MOVED,
String.valueOf(numSubDirMoved));
- auditParams.put(AUDIT_PARAM_SUBFILES_MOVED,
String.valueOf(numSubFilesMoved));
- auditParams.put(AUDIT_PARAM_DIRS_DELETED_LIST, String.join(",",
deletedDirNames));
- auditParams.put(AUDIT_PARAM_SUBDIRS_MOVED_LIST, String.join(",",
subDirNames));
- auditParams.put(AUDIT_PARAM_SUBFILES_MOVED_LIST, String.join(",",
subFileNames));
-
AUDIT.logWriteSuccess(ozoneManager.buildAuditMessageForSuccess(OMSystemAction.DIRECTORY_DELETION,
auditParams));
+ logDirectoryDeletionAuditSuccess(ozoneManager, fromSnapshotInfo,
numSubDirMoved, result);
}
} catch (IOException ex) {
// Case of IOException for fromProtobuf will not happen
@@ -264,69 +179,420 @@ public OMClientResponse
validateAndUpdateCache(OzoneManager ozoneManager, Execut
}
throw new IllegalStateException(ex);
} finally {
+ if (lockAcquired) {
+
mergeOmLockDetails(omMetadataManager.getLock().releaseWriteLocks(BUCKET_LOCK,
bucketLockKeys));
+ }
+ // Snapshot the mutated bucket infos for the response after releasing
the lock. The single apply thread is the
+ // only writer, so no other transaction can mutate them between release
and copy.
for (Map.Entry<Pair<String, String>, OmBucketInfo> entry :
- volBucketInfoMap.entrySet()) {
+ result.volBucketInfoMap.entrySet()) {
entry.setValue(entry.getValue().copyObject());
}
- if (lockAcquired) {
-
mergeOmLockDetails(omMetadataManager.getLock().releaseWriteLocks(BUCKET_LOCK,
bucketLockKeys));
- }
}
return new OMDirectoriesPurgeResponseWithFSO(
omResponse.build(), purgeRequests,
- getBucketLayout(), volBucketInfoMap, fromSnapshotInfo, openKeyInfoMap);
+ getBucketLayout(), result.volBucketInfoMap, fromSnapshotInfo,
result.openKeyInfoMap);
+ }
+
+ /**
+ * De-duplicates purged directories from the moved sub-directory set and
updates the deletion service metrics for
+ * this transaction, returning the resulting number of sub-directories moved
(used for the success audit).
+ */
+ private int recordDeletionMetrics(DeletingServiceMetrics
deletingServiceMetrics, OMMetrics omMetrics,
+ long numKeysProcessed, PurgeApplyResult result) {
+ // Apply the per-entry global OM metric mutations in bulk (one increment
each instead of once per entry) to
+ // minimize work done while holding the bucket write lock.
+ omMetrics.decNumKeys(numKeysProcessed);
+ omMetrics.incNumKeyDeletesInternal(numKeysProcessed);
+ // Remove deletedDirNames from subDirNames to avoid duplication
+ result.subDirNames.removeAll(result.deletedDirNames);
+ int numSubDirMoved = result.subDirNames.size();
+ deletingServiceMetrics.incrNumSubDirectoriesMoved(numSubDirMoved);
+ deletingServiceMetrics.incrNumSubFilesMoved(result.numSubFilesMoved);
+ deletingServiceMetrics.incrNumDirPurged(result.numDirsDeleted);
+ return numSubDirMoved;
+ }
+
+ /**
+ * Builds and emits the success audit message for a directory purge. Kept
separate so the per-entry parameter map is
+ * only assembled when debug audit logging is enabled.
+ */
+ private void logDirectoryDeletionAuditSuccess(OzoneManager ozoneManager,
SnapshotInfo fromSnapshotInfo,
+ int numSubDirMoved, PurgeApplyResult result) {
+ Map<String, String> auditParams = new LinkedHashMap<>();
+ if (fromSnapshotInfo != null) {
+ auditParams.put(AUDIT_PARAM_SNAPSHOT_ID,
fromSnapshotInfo.getSnapshotId().toString());
+ }
+ auditParams.put(AUDIT_PARAM_DIRS_DELETED,
String.valueOf(result.numDirsDeleted));
+ auditParams.put(AUDIT_PARAM_SUBDIRS_MOVED, String.valueOf(numSubDirMoved));
+ auditParams.put(AUDIT_PARAM_SUBFILES_MOVED,
String.valueOf(result.numSubFilesMoved));
+ auditParams.put(AUDIT_PARAM_DIRS_DELETED_LIST, String.join(",",
result.deletedDirNames));
+ auditParams.put(AUDIT_PARAM_SUBDIRS_MOVED_LIST, String.join(",",
result.subDirNames));
+ auditParams.put(AUDIT_PARAM_SUBFILES_MOVED_LIST, String.join(",",
result.subFileNames));
+
AUDIT.logWriteSuccess(ozoneManager.buildAuditMessageForSuccess(OMSystemAction.DIRECTORY_DELETION,
auditParams));
+ }
+
+ /**
+ * Logs, audits and builds the error response for a failure while resolving
the {@code fromSnapshot}. Extracted so
+ * the {@link #validateAndUpdateCache} catch block reads as a single call.
+ */
+ private OMClientResponse directoryPurgeFailure(OzoneManager ozoneManager,
OMResponse.Builder omResponse,
+ IOException e) {
+ LOG.error("Error occurred while performing OMDirectoriesPurge. ", e);
+ if (LOG.isDebugEnabled()) {
+
AUDIT.logWriteFailure(ozoneManager.buildAuditMessageForFailure(OMSystemAction.DIRECTORY_DELETION,
null, e));
+ }
+ return new
OMDirectoriesPurgeResponseWithFSO(createErrorOMResponse(omResponse, e));
+ }
+
+ /**
+ * Resolves the {@code fromSnapshot} this purge runs against and, for
new-format requests, validates that the
+ * previous snapshot chain has not changed. Extracted so {@link
#validateAndUpdateCache} reads as setup → per-path
+ * apply → bookkeeping.
+ */
+ private SnapshotInfo resolveFromSnapshotInfo(OzoneManager ozoneManager,
OmMetadataManagerImpl omMetadataManager,
+ PurgeDirectoriesRequest purgeDirsRequest, String fromSnapshot) throws
IOException {
+ SnapshotInfo fromSnapshotInfo = fromSnapshot != null
+ ? SnapshotUtils.getSnapshotInfo(ozoneManager, fromSnapshot) : null;
+ // Checking if this request is an old request or new one.
+ if (purgeDirsRequest.hasExpectedPreviousSnapshotID()) {
+ // Validating previous snapshot since while purging deletes, a snapshot
create request could make this purge
+ // directory request invalid on AOS since the deletedDirectory would be
in the newly created snapshot. Adding
+ // subdirectories could lead to not being able to reclaim sub-files and
subdirectories since the
+ // file/directory would be present in the newly created snapshot.
+ // Validating previous snapshot can ensure the chain hasn't changed.
+ UUID expectedPreviousSnapshotId =
purgeDirsRequest.getExpectedPreviousSnapshotID().hasUuid()
+ ?
fromProtobuf(purgeDirsRequest.getExpectedPreviousSnapshotID().getUuid()) : null;
+ validatePreviousSnapshotId(fromSnapshotInfo,
omMetadataManager.getSnapshotChainManager(),
+ expectedPreviousSnapshotId);
+ }
+ return fromSnapshotInfo;
+ }
+
+ /**
+ * Phase 1 (no lock): parses a path's sub-directories into prepared entries,
precomputing each delete/path key. The
+ * directory-table tombstones and bucket namespace decrements are applied
later under the lock in
+ * {@link #applyPreparedEntries}.
+ */
+ private List<PreparedEntry>
prepareMarkDeletedSubDirs(OzoneManagerProtocolProtos.PurgePathRequest path,
+ OmMetadataManagerImpl omMetadataManager) {
+ List<PreparedEntry> preparedSubDirs = new ArrayList<>();
+ for (OzoneManagerProtocolProtos.KeyInfo key :
path.getMarkDeletedSubDirsList()) {
+ ProcessedKeyInfo processed = processDeleteKey(key, path,
omMetadataManager);
+ preparedSubDirs.add(new PreparedEntry(processed, path.getBucketId(), 0L,
null));
+ }
+ return preparedSubDirs;
}
/**
- * Return the copy this request accumulates deltas on, fetching it on first
use. All three purge
- * paths can touch the same bucket, so they must share one copy or earlier
deltas are lost.
+ * Phase 1 (no lock): parses a path's sub-files into prepared entries,
precomputing each delete/path key, its
+ * replicated size and, for hsync files, the open-key name to clean up. The
file-table tombstones, hsync open-key
+ * cleanup and bucket quota decrements are applied later under the lock in
{@link #applyPreparedEntries}.
*/
- private OmBucketInfo accumulatedBucketInfo(Map<Pair<String, String>,
OmBucketInfo> volBucketInfoMap,
- OMMetadataManager omMetadataManager, String volumeName, String
bucketName) {
- OmBucketInfo omBucketInfo = volBucketInfoMap.get(Pair.of(volumeName,
bucketName));
+ private List<PreparedEntry>
prepareMoveDeletedSubFiles(OzoneManagerProtocolProtos.PurgePathRequest path,
+ OmMetadataManagerImpl omMetadataManager) {
+ List<PreparedEntry> preparedSubFiles = new ArrayList<>();
+ for (OzoneManagerProtocolProtos.KeyInfo key :
path.getDeletedSubFilesList()) {
+ ProcessedKeyInfo processed = processDeleteKey(key, path,
omMetadataManager);
+ long replicatedSize = sumBlockLengths(key);
+ // If omKeyInfo has hsync metadata, its corresponding open key is
cleaned up under the lock.
+ String hsyncClientId = getHsyncClientId(key);
+ String dbOpenKey = hsyncClientId == null ? null
+ : omMetadataManager.getOpenFileName(path.getVolumeId(),
path.getBucketId(),
+ processed.parentObjectID, processed.fileName, hsyncClientId);
+ preparedSubFiles.add(new PreparedEntry(processed, path.getBucketId(),
replicatedSize, dbOpenKey));
+ }
+ return preparedSubFiles;
+ }
- return omBucketInfo != null ? omBucketInfo
- : getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
+ /**
+ * Phase 1 (no lock): records a path's deleted directory for later purge
under the lock in
+ * {@link #applyPreparedEntries}.
+ */
+ private PreparedDirPurge
prepareDeletedDir(OzoneManagerProtocolProtos.PurgePathRequest path,
+ BucketNameInfo bucketNameInfo) {
+ return new PreparedDirPurge(bucketNameInfo.getVolumeName(),
bucketNameInfo.getBucketName(),
+ path.getBucketId(), path.getDeletedDir());
+ }
+
+ /**
+ * Phase 2 (under the bucket write lock): applies the prepared
directory/file tombstones, hsync open-key cleanup and
+ * per-bucket quota changes. Quota deltas are accumulated per bucket and
applied once, instead of once per entry, to
+ * minimize the mutations done while holding the write lock.
+ */
+ private void applyPreparedEntries(List<PreparedEntry> preparedSubDirs,
List<PreparedEntry> preparedSubFiles,
+ List<PreparedDirPurge> preparedDirPurges, OmMetadataManagerImpl
omMetadataManager, long trxnLogIndex,
+ PurgeApplyResult result) throws IOException {
+ // Holds one mutable bucket-info copy per bucket for this apply so that a
purge transaction touching many keys of
+ // the same bucket resolves it once and accumulates every quota delta on
the same copy.
+ Map<Pair<String, String>, OmBucketInfo> bucketInfoCache = new HashMap<>();
+ // Quota deltas are accumulated per bucket across all three entry loops
and applied once at the end.
+ Map<Pair<String, String>, QuotaDelta> quotaDeltas = new HashMap<>();
+
+ applyMarkedSubDirs(preparedSubDirs, omMetadataManager, trxnLogIndex,
result, bucketInfoCache, quotaDeltas);
+ applyMovedSubFiles(preparedSubFiles, omMetadataManager, trxnLogIndex,
result, bucketInfoCache, quotaDeltas);
+ applyDirPurges(preparedDirPurges, omMetadataManager, result,
bucketInfoCache, quotaDeltas);
+ applyQuotaDeltas(quotaDeltas);
+ }
+
+ /**
+ * Phase 2 (under the bucket write lock): tombstones each prepared
sub-directory in the directory table and
+ * accumulates its namespace quota decrement, when its bucket still matches
the prepared bucket id.
+ */
+ private void applyMarkedSubDirs(List<PreparedEntry> preparedSubDirs,
OmMetadataManagerImpl omMetadataManager,
+ long trxnLogIndex, PurgeApplyResult result, Map<Pair<String, String>,
OmBucketInfo> bucketInfoCache,
+ Map<Pair<String, String>, QuotaDelta> quotaDeltas) {
+ for (PreparedEntry entry : preparedSubDirs) {
+ ProcessedKeyInfo processed = entry.processed;
+ result.subDirNames.add(processed.deleteKey);
+ OmBucketInfo omBucketInfo = getBucketInfoCached(omMetadataManager,
+ bucketInfoCache, processed.volumeName, processed.bucketName);
+ // bucketInfo can be null in case of delete volume or bucket
+ // or key does not belong to bucket as bucket is recreated
+ if (null != omBucketInfo && omBucketInfo.getObjectID() ==
entry.bucketId) {
+ omMetadataManager.getDirectoryTable().addCacheEntry(new
CacheKey<>(processed.pathKey),
+ CacheValue.get(trxnLogIndex));
+ result.volBucketInfoMap.putIfAbsent(processed.volBucketPair,
omBucketInfo);
+ quotaDeltas.computeIfAbsent(processed.volBucketPair, k -> new
QuotaDelta(omBucketInfo)).usedNamespace += 1L;
+ }
+ }
+ }
+
+ /**
+ * Phase 2 (under the bucket write lock): tombstones each prepared sub-file
in the file table, cleans up its hsync
+ * open key when present, and accumulates its byte and namespace quota
decrements, when its bucket still matches the
+ * prepared bucket id.
+ */
+ private void applyMovedSubFiles(List<PreparedEntry> preparedSubFiles,
OmMetadataManagerImpl omMetadataManager,
+ long trxnLogIndex, PurgeApplyResult result, Map<Pair<String, String>,
OmBucketInfo> bucketInfoCache,
+ Map<Pair<String, String>, QuotaDelta> quotaDeltas) throws IOException {
+ for (PreparedEntry entry : preparedSubFiles) {
+ ProcessedKeyInfo processed = entry.processed;
+ result.subFileNames.add(processed.deleteKey);
+
+ // If omKeyInfo has hsync metadata, delete its corresponding open key as
well
+ if (entry.dbOpenKey != null) {
+ OmKeyInfo openKeyInfo =
omMetadataManager.getOpenKeyTable(getBucketLayout()).get(entry.dbOpenKey);
+ if (openKeyInfo != null) {
+ openKeyInfo = openKeyInfo.withMetadataMutations(
+ metadata -> metadata.put(DELETED_HSYNC_KEY, "true"));
+ result.openKeyInfoMap.put(entry.dbOpenKey, openKeyInfo);
+ }
+ }
+
+ result.numSubFilesMoved++;
+ OmBucketInfo omBucketInfo = getBucketInfoCached(omMetadataManager,
+ bucketInfoCache, processed.volumeName, processed.bucketName);
+ // bucketInfo can be null in case of delete volume or bucket
+ // or key does not belong to bucket as bucket is recreated
+ if (null != omBucketInfo && omBucketInfo.getObjectID() ==
entry.bucketId) {
+ omMetadataManager.getFileTable().addCacheEntry(new
CacheKey<>(processed.pathKey),
+ CacheValue.get(trxnLogIndex));
+ result.volBucketInfoMap.putIfAbsent(processed.volBucketPair,
omBucketInfo);
+ QuotaDelta quotaDelta =
quotaDeltas.computeIfAbsent(processed.volBucketPair, k -> new
QuotaDelta(omBucketInfo));
+ quotaDelta.usedBytes += entry.replicatedSize;
+ quotaDelta.usedNamespace += 1L;
+ }
+ }
+ }
+
+ /**
+ * Phase 2 (under the bucket write lock): records each purged directory and
accumulates its snapshot-namespace purge,
+ * when its bucket still matches the prepared bucket id.
+ */
+ private void applyDirPurges(List<PreparedDirPurge> preparedDirPurges,
OmMetadataManagerImpl omMetadataManager,
+ PurgeApplyResult result, Map<Pair<String, String>, OmBucketInfo>
bucketInfoCache,
+ Map<Pair<String, String>, QuotaDelta> quotaDeltas) {
+ for (PreparedDirPurge dirPurge : preparedDirPurges) {
+ result.deletedDirNames.add(dirPurge.deletedDir);
+ OmBucketInfo omBucketInfo = getBucketInfoCached(omMetadataManager,
+ bucketInfoCache, dirPurge.volumeName, dirPurge.bucketName);
+ if (omBucketInfo != null && omBucketInfo.getObjectID() ==
dirPurge.bucketId) {
+ Pair<String, String> volBucketPair =
Pair.of(omBucketInfo.getVolumeName(), omBucketInfo.getBucketName());
+ result.volBucketInfoMap.put(volBucketPair, omBucketInfo);
+ quotaDeltas.computeIfAbsent(volBucketPair, k -> new
QuotaDelta(omBucketInfo)).snapshotNamespacePurge += 1L;
+ }
+ result.numDirsDeleted++;
+ }
+ }
+
+ /**
+ * Phase 2 (under the bucket write lock): applies the per-bucket accumulated
quota changes once each, instead of once
+ * per entry, to minimize the mutations done while holding the bucket write
lock.
+ */
+ private void applyQuotaDeltas(Map<Pair<String, String>, QuotaDelta>
quotaDeltas) {
+ for (QuotaDelta quotaDelta : quotaDeltas.values()) {
+ if (quotaDelta.usedBytes != 0L) {
+ quotaDelta.bucketInfo.decrUsedBytes(quotaDelta.usedBytes, true);
+ }
+ if (quotaDelta.usedNamespace != 0L) {
+ quotaDelta.bucketInfo.decrUsedNamespace(quotaDelta.usedNamespace,
true);
+ }
+ if (quotaDelta.snapshotNamespacePurge != 0L) {
+
quotaDelta.bucketInfo.purgeSnapshotUsedNamespace(quotaDelta.snapshotNamespacePurge);
+ }
+ }
+ }
+
+ /**
+ * Phase 2 output accumulated by {@link #applyPreparedEntries} under the
bucket write lock: the mutated bucket infos
+ * and hsync open keys used to build the response, plus the moved
sub-dir/sub-file/deleted-dir names and counts used
+ * for the deletion metrics and the success audit.
+ */
+ private static final class PurgeApplyResult {
+ private final Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap =
new HashMap<>();
+ private final Map<String, OmKeyInfo> openKeyInfoMap = new HashMap<>();
+ private final Set<String> subDirNames = new HashSet<>();
+ private final Set<String> subFileNames = new HashSet<>();
+ private final Set<String> deletedDirNames = new HashSet<>();
+ private int numSubFilesMoved;
+ private int numDirsDeleted;
}
/**
* Helper class to hold processed key information.
*/
private static class ProcessedKeyInfo {
- private final OmKeyInfo keyInfo;
private final String deleteKey;
+ private final String pathKey;
private final String volumeName;
private final String bucketName;
+ private final long parentObjectID;
+ private final String fileName;
private final Pair<String, String> volBucketPair;
- ProcessedKeyInfo(OmKeyInfo keyInfo, String deleteKey, String volumeName,
String bucketName) {
- this.keyInfo = keyInfo;
+ ProcessedKeyInfo(String deleteKey, String pathKey, String volumeName,
+ String bucketName, long parentObjectID, String fileName) {
this.deleteKey = deleteKey;
+ this.pathKey = pathKey;
this.volumeName = volumeName;
this.bucketName = bucketName;
+ this.parentObjectID = parentObjectID;
+ this.fileName = fileName;
this.volBucketPair = Pair.of(volumeName, bucketName);
}
}
+ /**
+ * A sub-directory or sub-file prepared (lock-free) in phase 1 for
application under the bucket write lock in phase
+ * 2. {@code replicatedSize} is the file's replicated byte usage (0 for
directories) and {@code dbOpenKey} is the
+ * hsync open-key to clean up, or {@code null} when the entry is not an
hsync file.
+ */
+ private static final class PreparedEntry {
+ private final ProcessedKeyInfo processed;
+ private final long bucketId;
+ private final long replicatedSize;
+ private final String dbOpenKey;
+
+ PreparedEntry(ProcessedKeyInfo processed, long bucketId, long
replicatedSize, String dbOpenKey) {
+ this.processed = processed;
+ this.bucketId = bucketId;
+ this.replicatedSize = replicatedSize;
+ this.dbOpenKey = dbOpenKey;
+ }
+ }
+
+ /**
+ * A deleted directory prepared (lock-free) in phase 1 for its
snapshot-namespace purge under the bucket write lock
+ * in phase 2.
+ */
+ private static final class PreparedDirPurge {
+ private final String volumeName;
+ private final String bucketName;
+ private final long bucketId;
+ private final String deletedDir;
+
+ PreparedDirPurge(String volumeName, String bucketName, long bucketId,
String deletedDir) {
+ this.volumeName = volumeName;
+ this.bucketName = bucketName;
+ this.bucketId = bucketId;
+ this.deletedDir = deletedDir;
+ }
+ }
+
+ /**
+ * Accumulates a single bucket's quota changes across all entries in a purge
so they can be applied once under the
+ * write lock instead of once per entry.
+ */
+ private static final class QuotaDelta {
+ private final OmBucketInfo bucketInfo;
+ private long usedBytes;
+ private long usedNamespace;
+ private long snapshotNamespacePurge;
+
+ QuotaDelta(OmBucketInfo bucketInfo) {
+ this.bucketInfo = bucketInfo;
+ }
+ }
+
/**
* Process delete key info.
- * Returns ProcessedKeyInfo containing all the processed information.
+ * Reads only the fields the purge apply path needs directly from the
protobuf, instead of building a full
+ * {@link OmKeyInfo} (with its key-location, ACL, tag, encryption and
checksum objects) for every entry on the
+ * single-threaded apply path.
*/
private ProcessedKeyInfo processDeleteKey(OzoneManagerProtocolProtos.KeyInfo
key,
OzoneManagerProtocolProtos.PurgePathRequest path,
OmMetadataManagerImpl
omMetadataManager) {
- OmKeyInfo keyInfo = OmKeyInfo.getFromProtobuf(key);
+ long objectID = key.hasObjectID() ? key.getObjectID() : 0L;
+ long parentObjectID = key.hasParentID() ? key.getParentID() : 0L;
+ String fileName = OzoneFSUtils.getFileName(key.getKeyName());
String pathKey = omMetadataManager.getOzonePathKey(path.getVolumeId(),
- path.getBucketId(), keyInfo.getParentObjectID(),
keyInfo.getFileName());
- String deleteKey = omMetadataManager.getOzoneDeletePathKey(
- keyInfo.getObjectID(), pathKey);
+ path.getBucketId(), parentObjectID, fileName);
+ String deleteKey = omMetadataManager.getOzoneDeletePathKey(objectID,
pathKey);
+
+ return new ProcessedKeyInfo(deleteKey, pathKey, key.getVolumeName(),
key.getBucketName(),
+ parentObjectID, fileName);
+ }
- String volumeName = keyInfo.getVolumeName();
- String bucketName = keyInfo.getBucketName();
+ /**
+ * Returns a single mutable bucket-info copy for the given volume/bucket,
memoized within one apply so that all three
+ * purge paths touching the same bucket share one copy (or earlier quota
deltas would be lost) and resolve the bucket
+ * cache lookup once. The copy comes from {@link #getBucketInfoForUpdate},
so quota mutations stay off the live cache
+ * until they are published on the success path. {@code null} results
(deleted bucket) are memoized too.
+ */
+ private static OmBucketInfo getBucketInfoCached(OmMetadataManagerImpl
omMetadataManager,
+ Map<Pair<String, String>, OmBucketInfo> cache, String volumeName, String
bucketName) {
+ Pair<String, String> cacheKey = Pair.of(volumeName, bucketName);
+ if (cache.containsKey(cacheKey)) {
+ return cache.get(cacheKey);
+ }
+ OmBucketInfo omBucketInfo = getBucketInfoForUpdate(omMetadataManager,
volumeName, bucketName);
+ cache.put(cacheKey, omBucketInfo);
+ return omBucketInfo;
+ }
- return new ProcessedKeyInfo(keyInfo, deleteKey, volumeName, bucketName);
+ /**
+ * Reads the HSYNC client id directly from the key's protobuf metadata,
mirroring
+ * {@code KeyValueUtil.getFromProtobuf(...).get(HSYNC_CLIENT_ID)} without
building the full metadata map.
+ */
+ private static String getHsyncClientId(OzoneManagerProtocolProtos.KeyInfo
key) {
+ for (HddsProtos.KeyValue kv : key.getMetadataList()) {
+ if (OzoneConsts.HSYNC_CLIENT_ID.equals(kv.getKey())) {
+ return kv.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Computes replicated byte usage for a file directly from its protobuf key
locations. This is equivalent to
+ * {@link OMKeyRequest#sumBlockLengths(OmKeyInfo)} on the parsed key: {@code
getFromProtobuf} only regroups the same
+ * locations by create-version, so summing every {@code KeyLocation} length
yields the identical total while avoiding
+ * the full OmKeyInfo build on the apply path.
+ */
+ private static long sumBlockLengths(OzoneManagerProtocolProtos.KeyInfo key) {
+ ReplicationConfig replicationConfig =
ReplicationConfig.fromProto(key.getType(), key.getFactor(),
+ key.getEcReplicationConfig());
+ long bytesUsed = 0;
+ for (OzoneManagerProtocolProtos.KeyLocationList group :
key.getKeyLocationListList()) {
+ for (OzoneManagerProtocolProtos.KeyLocation location :
group.getKeyLocationsList()) {
+ bytesUsed += QuotaUtil.getReplicatedSize(location.getLength(),
replicationConfig);
+ }
+ }
+ return bytesUsed;
}
private List<String[]> getBucketLockKeySet(PurgeDirectoriesRequest
purgeDirsRequest) {
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java
index 9b3f2b2a7a3..ef3bd91a3e2 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java
@@ -34,6 +34,7 @@
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -536,38 +537,49 @@ private OzoneManagerProtocolProtos.PurgePathRequest
wrapPurgeRequest(
private List<OzoneManagerProtocolProtos.OMResponse>
submitPurgePathsWithBatching(List<PurgePathRequest> requests,
String snapTableKey, UUID expectedPreviousSnapshotId,
Map<VolumeBucketId, BucketNameInfo> bucketNameInfoMap) {
+ // Group purge paths by their owning bucket so that every submitted purge
transaction contains paths from a single
+ // bucket only. This keeps the apply side acquiring exactly one bucket
write lock per transaction, so a large
+ // background directory purge cannot block readers on other buckets.
+ Map<VolumeBucketId, List<PurgePathRequest>> requestsByBucket = new
LinkedHashMap<>();
+ for (PurgePathRequest req : requests) {
+ requestsByBucket.computeIfAbsent(new VolumeBucketId(req.getVolumeId(),
req.getBucketId()),
+ k -> new ArrayList<>()).add(req);
+ }
+
List<OzoneManagerProtocolProtos.OMResponse> responses = new ArrayList<>();
- List<PurgePathRequest> purgePathRequestBatch = new ArrayList<>();
- long batchBytes = 0;
+ for (List<PurgePathRequest> bucketRequests : requestsByBucket.values()) {
+ List<PurgePathRequest> purgePathRequestBatch = new ArrayList<>();
+ long batchBytes = 0;
+
+ for (PurgePathRequest req : bucketRequests) {
+ int reqSize = req.getSerializedSize();
+
+ // If adding this request would exceed the limit, flush the current
batch first
+ if (batchBytes + reqSize > ratisByteLimit &&
!purgePathRequestBatch.isEmpty()) {
+ OzoneManagerProtocolProtos.OMResponse resp =
+ submitPurgeRequest(snapTableKey, expectedPreviousSnapshotId,
bucketNameInfoMap, purgePathRequestBatch);
+ if (resp == null || !resp.getSuccess()) {
+ return Collections.emptyList();
+ }
+ responses.add(resp);
+ purgePathRequestBatch.clear();
+ batchBytes = 0;
+ }
- for (PurgePathRequest req : requests) {
- int reqSize = req.getSerializedSize();
+ // Add current request to batch
+ purgePathRequestBatch.add(req);
+ batchBytes += reqSize;
+ }
- // If adding this request would exceed the limit, flush the current
batch first
- if (batchBytes + reqSize > ratisByteLimit &&
!purgePathRequestBatch.isEmpty()) {
+ // Flush remaining batch for this bucket if any
+ if (!purgePathRequestBatch.isEmpty()) {
OzoneManagerProtocolProtos.OMResponse resp =
submitPurgeRequest(snapTableKey, expectedPreviousSnapshotId,
bucketNameInfoMap, purgePathRequestBatch);
- if (!resp.getSuccess()) {
+ if (resp == null || !resp.getSuccess()) {
return Collections.emptyList();
}
responses.add(resp);
- purgePathRequestBatch.clear();
- batchBytes = 0;
- }
-
- // Add current request to batch
- purgePathRequestBatch.add(req);
- batchBytes += reqSize;
- }
-
- // Flush remaining batch if any
- if (!purgePathRequestBatch.isEmpty()) {
- OzoneManagerProtocolProtos.OMResponse resp =
- submitPurgeRequest(snapTableKey, expectedPreviousSnapshotId,
bucketNameInfoMap, purgePathRequestBatch);
- if (!resp.getSuccess()) {
- return Collections.emptyList();
}
- responses.add(resp);
}
return responses;
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeApplyPerf.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeApplyPerf.java
new file mode 100644
index 00000000000..0debbb9e7e7
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeApplyPerf.java
@@ -0,0 +1,226 @@
+/*
+ * 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.request.key;
+
+import static
org.apache.hadoop.ozone.om.request.file.OMFileRequest.getOmKeyInfo;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
+import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
+import org.apache.hadoop.ozone.ClientVersion;
+import org.apache.hadoop.ozone.om.execution.flowcontrol.ExecutionContext;
+import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
+import org.apache.hadoop.ozone.om.request.OMRequestTestUtils;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BucketNameInfo;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PurgePathRequest;
+import org.apache.hadoop.util.Time;
+import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ratis.server.protocol.TermIndex;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.event.Level;
+
+/**
+ * Handler-level microbenchmark for {@link
OMDirectoriesPurgeRequestWithFSO#validateAndUpdateCache}.
+ *
+ * <p>The OM applies write transactions on a single serial apply thread, so
the wall time of
+ * {@code validateAndUpdateCache} directly bounds how long all other
writes/renames/mkdirs wait behind a background
+ * directory purge. This benchmark times that apply call over a single large
single-bucket {@code PurgeDirectories}
+ * transaction, isolating the apply-thread CPU that the workspace changes
reduce:
+ * <ul>
+ * <li>sub-entries are read as scalars straight from the protobuf instead of
building a full {@link OmKeyInfo}
+ * (with its key-location, ACL, metadata and encryption objects) per
entry, and</li>
+ * <li>the owning bucket's cache entry is memoized once per apply instead of
looked up per entry.</li>
+ * </ul>
+ *
+ * <p>The {@code benchmark} tag is excluded from {@code mvn test} and CI by
default, so it must be re-enabled
+ * explicitly to run on demand:
+ * <pre>
+ * mvn -pl :ozone-manager test -DskipShade -DskipRecon -DskipDocs \
+ * -Dtest=TestOMDirectoriesPurgeApplyPerf -Dgroups=benchmark
-Dexcluded-test-groups= \
+ * -Dsurefire.failIfNoSpecifiedTests=false
+ * </pre>
+ * Tunables (system properties): {@code bench.subEntries} (default 1000),
{@code bench.iters} (default 20),
+ * {@code bench.warmup} (default 5).
+ */
+@Tag("benchmark")
+public class TestOMDirectoriesPurgeApplyPerf extends OMKeyRequestTests {
+
+ private static final long DIR_OBJECT_ID = 1L;
+ private static final long BLOCK_LENGTH = 100L;
+ private static final int VERSIONS_PER_FILE = 2;
+ private static final int BLOCKS_PER_VERSION = 3;
+
+ @Test
+ public void benchmarkApplyThreadTime() throws Exception {
+ // Match production: the debug-guarded success audit is off by default, so
it should not colour the timing.
+ GenericTestUtils.setLogLevel(OMKeyRequest.class, Level.INFO);
+
+ final int subEntries = Integer.getInteger("bench.subEntries", 1000);
+ final int iters = Integer.getInteger("bench.iters", 20);
+ final int warmup = Integer.getInteger("bench.warmup", 5);
+ final long applies = (long) (iters + warmup);
+
+ when(ozoneManager.getDefaultReplicationConfig())
+
.thenReturn(RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE));
+
+ String bucket = "bucket" + UUID.randomUUID();
+ OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucket,
omMetadataManager,
+ BucketLayout.FILE_SYSTEM_OPTIMIZED);
+ String bucketKey = omMetadataManager.getBucketKey(volumeName, bucket);
+ OmBucketInfo bucketInfo =
omMetadataManager.getBucketTable().get(bucketKey);
+ long volumeId = omMetadataManager.getVolumeId(volumeName);
+ long bucketId = bucketInfo.getObjectID();
+
+ // Seed enough quota that the per-apply decrements (repeated across warmup
+ measured runs) never underflow.
+ long perFileReplicatedBytes = replicatedBytesPerFile();
+ bucketInfo.incrUsedBytes(applies * subEntries * perFileReplicatedBytes *
2);
+ bucketInfo.incrUsedNamespace(applies * subEntries * 2L * 2);
+ omMetadataManager.getBucketTable().addCacheEntry(new
CacheKey<>(bucketKey), CacheValue.get(1L, bucketInfo));
+ omMetadataManager.getBucketTable().put(bucketKey, bucketInfo);
+
+ OMRequest preExecuted = preExecutePurgeRequest(
+ buildSingleBucketPurgeRequest(volumeId, bucketId, bucket, subEntries));
+
+ // Warmup (let the JIT compile the apply path) then measure.
+ for (int i = 0; i < warmup; i++) {
+ applyOnce(preExecuted);
+ }
+ long[] samplesNs = new long[iters];
+ long blackhole = 0;
+ for (int i = 0; i < iters; i++) {
+ long t0 = System.nanoTime();
+ blackhole += applyOnce(preExecuted);
+ samplesNs[i] = System.nanoTime() - t0;
+ }
+
+ Arrays.sort(samplesNs);
+ long sum = 0;
+ for (long s : samplesNs) {
+ sum += s;
+ }
+ double meanUs = sum / (double) iters / 1000.0;
+ double p50Us = samplesNs[iters / 2] / 1000.0;
+ double p90Us = samplesNs[Math.min(iters - 1, (int) Math.ceil(iters * 0.9)
- 1)] / 1000.0;
+ double minUs = samplesNs[0] / 1000.0;
+ double maxUs = samplesNs[iters - 1] / 1000.0;
+
+ // Single grep-able line so baseline and changed runs can be compared
directly.
+ System.out.printf("BENCH apply subEntries=%d (subFiles+subDirs) iters=%d
warmup=%d "
+ + "meanUs=%.1f p50Us=%.1f p90Us=%.1f minUs=%.1f maxUs=%.1f
totalMs=%.1f blackhole=%d%n",
+ subEntries, iters, warmup, meanUs, p50Us, p90Us, minUs, maxUs, sum /
1_000_000.0, blackhole);
+ }
+
+ private long applyOnce(OMRequest preExecuted) {
+ OMDirectoriesPurgeRequestWithFSO request = new
OMDirectoriesPurgeRequestWithFSO(preExecuted);
+ // Only validateAndUpdateCache runs on the apply thread; the RocksDB batch
flush happens off-thread later, so it is
+ // intentionally excluded here.
+ ExecutionContext context = ExecutionContext.of(100L, TermIndex.valueOf(1L,
100L));
+ // Consume the (always non-null) response as a blackhole so the JIT cannot
elide the apply call.
+ return
System.identityHashCode(request.validateAndUpdateCache(ozoneManager, context));
+ }
+
+ private long replicatedBytesPerFile() {
+ // getReplicatedSize(len, RATIS THREE) == 3 * len; VERSIONS_PER_FILE *
BLOCKS_PER_VERSION blocks per file.
+ return (long) VERSIONS_PER_FILE * BLOCKS_PER_VERSION * BLOCK_LENGTH * 3;
+ }
+
+ private OMRequest preExecutePurgeRequest(OMRequest omRequest) throws
java.io.IOException {
+ return new OMKeyPurgeRequest(omRequest).preExecute(ozoneManager);
+ }
+
+ /**
+ * Builds one {@link PurgePathRequest} that marks {@code subEntries}
sub-directories deleted and moves
+ * {@code subEntries} sub-files, all under a single bucket, wrapped in a
{@code PurgeDirectories} {@link OMRequest}.
+ */
+ private OMRequest buildSingleBucketPurgeRequest(long volumeId, long
bucketId, String bucket, int subEntries) {
+ PurgePathRequest.Builder path = PurgePathRequest.newBuilder()
+ .setVolumeId(volumeId)
+ .setBucketId(bucketId);
+
+ long objectId = 100L;
+ for (int i = 0; i < subEntries; i++) {
+ OmDirectoryInfo subdir = OmDirectoryInfo.newBuilder()
+ .setName("subdir" + i)
+ .setCreationTime(Time.now())
+ .setModificationTime(Time.now())
+ .setObjectID(objectId++)
+ .setParentObjectID(DIR_OBJECT_ID)
+ .setUpdateID(0)
+ .build();
+ OmKeyInfo subDirKey = getOmKeyInfo(volumeName, bucket, subdir, "dir1/" +
subdir.getName());
+
path.addMarkDeletedSubDirs(subDirKey.getProtobuf(ClientVersion.CURRENT_VERSION));
+
+ OmKeyInfo subFile = OMRequestTestUtils.createOmKeyInfo(volumeName,
bucket, "dir1/file" + i,
+
RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE))
+ .setObjectID(objectId++)
+ .setParentObjectID(DIR_OBJECT_ID)
+ .setUpdateID(100L)
+ .build();
+ subFile.setKeyLocationVersions(buildLocationVersions());
+ path.addDeletedSubFiles(subFile.getProtobuf(true,
ClientVersion.CURRENT_VERSION));
+ }
+
+ OzoneManagerProtocolProtos.PurgeDirectoriesRequest.Builder purgeDir =
+ OzoneManagerProtocolProtos.PurgeDirectoriesRequest.newBuilder()
+ .addDeletedPath(path.build())
+ .addBucketNameInfos(BucketNameInfo.newBuilder()
+ .setVolumeName(volumeName).setBucketName(bucket)
+ .setVolumeId(volumeId).setBucketId(bucketId).build());
+
+ return OMRequest.newBuilder()
+ .setCmdType(OzoneManagerProtocolProtos.Type.PurgeDirectories)
+ .setPurgeDirectoriesRequest(purgeDir)
+ .setClientId(UUID.randomUUID().toString())
+ .build();
+ }
+
+ private List<OmKeyLocationInfoGroup> buildLocationVersions() {
+ List<OmKeyLocationInfoGroup> versions = new ArrayList<>(VERSIONS_PER_FILE);
+ for (int v = 0; v < VERSIONS_PER_FILE; v++) {
+ List<OmKeyLocationInfo> locations = new ArrayList<>(BLOCKS_PER_VERSION);
+ for (int b = 0; b < BLOCKS_PER_VERSION; b++) {
+ locations.add(new OmKeyLocationInfo.Builder()
+ .setLength(BLOCK_LENGTH)
+ .setBlockID(new BlockID(1 + v, 1 + b))
+ .build());
+ }
+ versions.add(new OmKeyLocationInfoGroup(v, locations, false));
+ }
+ return versions;
+ }
+
+ @Override
+ public BucketLayout getBucketLayout() {
+ return BucketLayout.FILE_SYSTEM_OPTIMIZED;
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeRequestAndResponse.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeRequestAndResponse.java
index 8c74321f848..f5442acfdc7 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeRequestAndResponse.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeRequestAndResponse.java
@@ -56,6 +56,7 @@
import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
import org.apache.hadoop.ozone.ClientVersion;
+import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.OmSnapshot;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
@@ -618,6 +619,91 @@ public void
testValidateAndUpdateCacheQuotaBucketRecreated()
validateDeletedKeys(omMetadataManager, deletedKeyNames);
}
+ /**
+ * Guards the apply-path light parse in {@link
OMDirectoriesPurgeRequestWithFSO}: for a sub-file the request reads the
+ * replicated byte sum and the hsync client id directly from the protobuf
instead of building a full
+ * {@link OmKeyInfo}. This asserts the two subtle spots stay equivalent to
the full parse: a multi-version,
+ * replication-factor-three block sum decrements bucket quota to exactly
zero, and the matching open key is marked
+ * deleted via its hsync metadata.
+ */
+ @Test
+ public void testLightParsePreservesReplicatedByteSumAndHsyncForSubFiles()
throws Exception {
+ when(ozoneManager.getDefaultReplicationConfig())
+
.thenReturn(RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE));
+ String bucket = "bucket" + RandomUtils.secure().randomInt();
+ OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucket,
+ omMetadataManager, BucketLayout.FILE_SYSTEM_OPTIMIZED);
+ String bucketKey = omMetadataManager.getBucketKey(volumeName, bucket);
+ OmBucketInfo bucketInfo =
omMetadataManager.getBucketTable().get(bucketKey);
+ long volumeId = omMetadataManager.getVolumeId(volumeName);
+ long bucketId = bucketInfo.getObjectID();
+
+ OmDirectoryInfo dir1 = OmDirectoryInfo.newBuilder()
+ .setName("dir1")
+ .setCreationTime(Time.now())
+ .setModificationTime(Time.now())
+ .setObjectID(1)
+ .setParentObjectID(bucketId)
+ .setUpdateID(0)
+ .build();
+ OMRequestTestUtils.addDirKeyToDirTable(false, dir1, volumeName, bucket,
1L, omMetadataManager);
+
+ // Sub-file with RATIS THREE replication, two block-location versions of
different lengths, and hsync metadata.
+ String hsyncClientId = UUID.randomUUID().toString();
+ OmKeyInfo subFile = OMRequestTestUtils.createOmKeyInfo(volumeName, bucket,
"file1",
+
RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE))
+ .setObjectID(2L)
+ .setParentObjectID(dir1.getObjectID())
+ .setUpdateID(100L)
+ .addMetadata(OzoneConsts.HSYNC_CLIENT_ID, hsyncClientId)
+ .build();
+ List<OmKeyLocationInfoGroup> locationVersions = new ArrayList<>();
+ locationVersions.add(new OmKeyLocationInfoGroup(1,
Collections.singletonList(
+ new OmKeyLocationInfo.Builder().setLength(100L).setBlockID(new
BlockID(1, 1)).build()), false));
+ locationVersions.add(new OmKeyLocationInfoGroup(2,
Collections.singletonList(
+ new OmKeyLocationInfo.Builder().setLength(250L).setBlockID(new
BlockID(1, 2)).build()), false));
+ subFile.setKeyLocationVersions(locationVersions);
+
+ long expectedReplicatedBytes = OMKeyRequest.sumBlockLengths(subFile);
+ assertEquals((100L + 250L) * 3, expectedReplicatedBytes);
+
+ // Seed bucket quota to exactly the full-parse byte sum so a correct
light-parse decrement lands on zero.
+ bucketInfo.incrUsedBytes(expectedReplicatedBytes);
+ bucketInfo.incrUsedNamespace(1L);
+ omMetadataManager.getBucketTable().addCacheEntry(new
CacheKey<>(bucketKey), CacheValue.get(1L, bucketInfo));
+ omMetadataManager.getBucketTable().put(bucketKey, bucketInfo);
+
+ // Seed the matching open key that the hsync purge leg should mark deleted.
+ String dbOpenKey = omMetadataManager.getOpenFileName(volumeId, bucketId,
dir1.getObjectID(), "file1",
+ hsyncClientId);
+ OmKeyInfo openKeyInfo = OMRequestTestUtils.createOmKeyInfo(volumeName,
bucket, "file1",
+
RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE))
+ .setObjectID(2L)
+ .setParentObjectID(dir1.getObjectID())
+ .setUpdateID(100L)
+ .build();
+
omMetadataManager.getOpenKeyTable(BucketLayout.FILE_SYSTEM_OPTIMIZED).put(dbOpenKey,
openKeyInfo);
+
+ subFile.setKeyName("dir1/file1");
+ OMRequest omRequest = createPurgeKeysRequest(null, null,
Collections.emptyList(),
+ Collections.singletonList(subFile), bucketInfo);
+ OMDirectoriesPurgeRequestWithFSO purgeRequest =
+ new OMDirectoriesPurgeRequestWithFSO(preExecute(omRequest));
+ OMDirectoriesPurgeResponseWithFSO omClientResponse =
+ (OMDirectoriesPurgeResponseWithFSO)
purgeRequest.validateAndUpdateCache(ozoneManager, 100L);
+
+ // Light-parse block-length sum matches the full OmKeyInfo parse: bucket
usedBytes drops exactly to zero.
+ OmBucketInfo updatedBucketInfo =
omMetadataManager.getBucketTable().get(bucketKey);
+ assertEquals(0L, updatedBucketInfo.getUsedBytes());
+ assertEquals(expectedReplicatedBytes,
updatedBucketInfo.getSnapshotUsedBytes());
+
+ performBatchOperationCommit(omClientResponse);
+
+ // hsync leg: the corresponding open key is marked deleted after the batch
commit.
+ OmKeyInfo purgedOpenKey =
omMetadataManager.getOpenKeyTable(BucketLayout.FILE_SYSTEM_OPTIMIZED).get(dbOpenKey);
+ assertEquals("true",
purgedOpenKey.getMetadata().get(OzoneConsts.DELETED_HSYNC_KEY));
+ }
+
private void performBatchOperationCommit(OMDirectoriesPurgeResponseWithFSO
omClientResponse)
throws ExecutionException, InterruptedException {
CompletableFuture<Void> future = new CompletableFuture<>();
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingService.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingService.java
index fdb723dc319..4f3e2a2eeec 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingService.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingService.java
@@ -25,7 +25,6 @@
import java.io.File;
import java.io.IOException;
-import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
@@ -265,8 +264,7 @@ void testPurgeDirectoriesBatching() throws Exception {
final int ratisLimitBytes = 2304;
OzoneConfiguration conf = new OzoneConfiguration();
- File testDir = Files.createTempDirectory("TestDDS-SubmitSpy").toFile();
- ServerUtils.setOzoneMetaDirPath(conf, testDir.toString());
+ ServerUtils.setOzoneMetaDirPath(conf, folder.toFile().toString());
conf.setTimeDuration(OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL,
100, TimeUnit.MILLISECONDS);
conf.setStorageSize(OMConfigKeys.OZONE_OM_RATIS_LOG_APPENDER_QUEUE_BYTE_LIMIT,
ratisLimitBytes, StorageUnit.BYTES);
conf.setQuietMode(false);
@@ -327,8 +325,62 @@ void testPurgeDirectoriesBatching() throws Exception {
assertThat(payloadBytes).as("Batch size should respect Ratis byte
limit").isLessThanOrEqualTo(ratisLimitBytes);
}
+ }
+
+ @Test
+ @DisplayName("DirectoryDeletingService submits one bucket per
PurgeDirectories transaction")
+ void testPurgeDirectoriesGroupedByBucketPerTransaction() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ ServerUtils.setOzoneMetaDirPath(conf, folder.toFile().toString());
+ conf.setTimeDuration(OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL,
100, TimeUnit.MILLISECONDS);
+ conf.setQuietMode(false);
+
+ OmTestManagers managers = new OmTestManagers(conf);
+ om = managers.getOzoneManager();
+ KeyManager km = managers.getKeyManager();
+
+ DirectoryDeletingService real = (DirectoryDeletingService)
km.getDirDeletingService();
+ DirectoryDeletingService dds = Mockito.spy(real);
+
+ List<OzoneManagerProtocolProtos.OMRequest> captured = new ArrayList<>();
+ Mockito.doAnswer(inv -> {
+ captured.add(inv.getArgument(0));
+ return OzoneManagerProtocolProtos.OMResponse.newBuilder()
+
.setCmdType(OzoneManagerProtocolProtos.Type.PurgeDirectories).setStatus(OzoneManagerProtocolProtos.Status.OK)
+ .build();
+
}).when(dds).submitRequest(Mockito.any(OzoneManagerProtocolProtos.OMRequest.class));
+
+ final long volumeId = 1L, bucketIdA = 2L, bucketIdB = 3L;
+ // Interleave paths from two different buckets to prove grouping is
order-independent.
+ List<OzoneManagerProtocolProtos.PurgePathRequest> purgeList = new
ArrayList<>();
+ Map<org.apache.hadoop.ozone.om.OMMetadataManager.VolumeBucketId,
OzoneManagerProtocolProtos.BucketNameInfo>
+ bucketNameInfoMap = new HashMap<>();
+ for (long bucketId : new long[] {bucketIdA, bucketIdB}) {
+ bucketNameInfoMap.put(new
org.apache.hadoop.ozone.om.OMMetadataManager.VolumeBucketId(volumeId, bucketId),
+
OzoneManagerProtocolProtos.BucketNameInfo.newBuilder().setVolumeId(volumeId).setBucketId(bucketId)
+ .setVolumeName("v").setBucketName("b" + bucketId).build());
+ }
+ for (int i = 0; i < 6; i++) {
+ purgeList.add(OzoneManagerProtocolProtos.PurgePathRequest.newBuilder()
+ .setVolumeId(volumeId).setBucketId(bucketIdA).setDeletedDir("a-dir-"
+ i).build());
+ purgeList.add(OzoneManagerProtocolProtos.PurgePathRequest.newBuilder()
+ .setVolumeId(volumeId).setBucketId(bucketIdB).setDeletedDir("b-dir-"
+ i).build());
+ }
+
+ dds.optimizeDirDeletesAndSubmitRequest(0L, 0L, 0L, new ArrayList<>(),
purgeList, null, Time.monotonicNow(), km,
+ kv -> true, kv -> true, bucketNameInfoMap, null, 1L, new
AtomicInteger(Integer.MAX_VALUE));
- org.apache.commons.io.FileUtils.deleteDirectory(testDir);
+ assertThat(captured).as("Expect one transaction per bucket when the byte
limit is not hit").hasSize(2);
+ Set<Long> bucketsSeen = new HashSet<>();
+ for (OzoneManagerProtocolProtos.OMRequest omReq : captured) {
+ OzoneManagerProtocolProtos.PurgeDirectoriesRequest purge =
omReq.getPurgeDirectoriesRequest();
+ Set<Long> bucketsInTxn = purge.getDeletedPathList().stream()
+
.map(OzoneManagerProtocolProtos.PurgePathRequest::getBucketId).collect(java.util.stream.Collectors.toSet());
+ assertThat(bucketsInTxn).as("Each transaction must contain paths from a
single bucket").hasSize(1);
+ bucketsSeen.addAll(bucketsInTxn);
+ }
+ assertThat(bucketsSeen).as("Both buckets should be purged across
transactions").containsExactlyInAnyOrder(
+ bucketIdA, bucketIdB);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
