+ * --read-percent tunes the mix: every operation of the run independently draws + * whether to write a file or to read one back, so a run can be made read-heavy + * or write-heavy without changing what any one read validates. -n therefore + * counts operations of both kinds rather than writes alone, and the split holds + * on average rather than exactly. + *
* CRC32 keeps the validation off the critical path of the measured throughput.
* Successive writes of a path differ in the marker only, and markers less than
* 2^32 apart never share a CRC32, so a stale read is always detected.
@@ -89,6 +95,15 @@ public class HadoopFsReadWriteValidator extends HadoopBaseFreonGenerator
defaultValue = "10000")
private int maxFilesPerThread;
+ @Option(names = {"--read-percent"},
+ description = "Percentage of the operations that read a file back and validate it instead of writing one. "
+ + "0 only writes, 50 pairs a read with every write on average, and 90 makes the run read-heavy. Every "
+ + "operation draws on its own, so the split holds over a run rather than on any particular pair of "
+ + "operations. A read validates a file the thread wrote, picked at random; a thread that has written "
+ + "nothing yet has nothing to read and writes instead.",
+ defaultValue = "50")
+ private double readPercent;
+
private ContentGenerator contentGenerator;
private Timer writeTimer;
@@ -113,6 +128,11 @@ public Void call() throws Exception {
throw new IllegalArgumentException(
"--max-files-per-thread must be positive");
}
+ // negated so that NaN, which fails every comparison, is rejected too
+ if (!(readPercent >= 0 && readPercent <= 100)) {
+ throw new IllegalArgumentException(
+ "--read-percent must be between 0 and 100");
+ }
super.init();
@@ -129,7 +149,7 @@ public Void call() throws Exception {
writeTimer = getMetrics().timer("file-write");
readTimer = getMetrics().timer("file-read-validate");
- runTests(this::writeAndValidate);
+ runTests(this::readOrWrite);
} finally {
org.apache.hadoop.hdds.utils.IOUtils.closeQuietly(fileSystem);
}
@@ -137,11 +157,29 @@ public Void call() throws Exception {
return null;
}
- private void writeAndValidate(long counter) throws Exception {
+ /**
+ * One operation of the run, a read or a write, drawn per --read-percent. A
+ * thread that has written nothing yet has nothing to read back and writes.
+ */
+ private void readOrWrite(long counter) throws Exception {
ThreadHistory history = threadHistory.get();
+ if (history.isEmpty() || !readsNext()) {
+ writeAndRecord(history, counter);
+ } else {
+ validateRandomFile(history);
+ }
+ }
+
+ /** Draws whether this operation reads, see --read-percent. */
+ private boolean readsNext() {
+ return ThreadLocalRandom.current().nextDouble(100) < readPercent;
+ }
+
+ private void writeAndRecord(ThreadHistory history, long counter)
+ throws Exception {
+ long marker = history.nextMarker();
long fileId = counter % maxFilesPerThread;
Path file = objectPath(fileId);
- long marker = history.nextMarker();
long checksum;
try {
@@ -154,7 +192,14 @@ private void writeAndValidate(long counter) throws Exception {
throw e;
}
history.record(fileId, checksum);
+ }
+ /**
+ * Read back one of the files this thread wrote, picked at random, and verify
+ * that its content still matches the checksum of the latest write of that
+ * path.
+ */
+ private void validateRandomFile(ThreadHistory history) throws Exception {
long readId = history.randomFileId();
Path target = objectPath(readId);
long expected = history.checksumOf(readId);
@@ -168,10 +213,13 @@ private void writeAndValidate(long counter) throws Exception {
}
/**
- * Path of the file for the given counter. The thread sequence id is part of
- * the path so each worker owns a private namespace; paths are reused once a
- * thread has written --max-files-per-thread of them, and this keeps one
- * thread from overwriting a file another thread is reading back.
+ * Path of the file for the given id. The thread sequence id is part of the
+ * path so each worker owns a private namespace, which keeps one thread from
+ * overwriting a file another thread is reading back. The id comes from the
+ * task counter, which the framework recycles within -n, so a time-based run
+ * writes over its paths rather than growing without bound; the modulo caps a
+ * count-based run at --max-files-per-thread distinct ids, and with it the
+ * checksums a thread has to remember.
*/
private Path objectPath(long fileId) {
return new Path(getRootPath() + "/" + generateObjectName(fileId)
@@ -252,6 +300,11 @@ private void forget(long fileId) {
}
}
+ /** Whether the thread has written anything it could read back yet. */
+ private boolean isEmpty() {
+ return fileIds.isEmpty();
+ }
+
private long randomFileId() {
return fileIds.get(ThreadLocalRandom.current().nextInt(fileIds.size()));
}
diff --git a/hadoop-ozone/freon/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopFsReadPercent.java b/hadoop-ozone/freon/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopFsReadPercent.java
new file mode 100644
index 000000000000..3ab3bfacde54
--- /dev/null
+++ b/hadoop-ozone/freon/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopFsReadPercent.java
@@ -0,0 +1,280 @@
+/*
+ * 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.freon;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.within;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSInputStream;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import picocli.CommandLine;
+
+/**
+ * Verifies that {@code --read-percent} of {@link HadoopFsReadWriteValidator}
+ * (dfsrw) controls how the operations of a run split between reads and writes.
+ * The workload only needs a Hadoop {@code FileSystem}, so these run against the
+ * local one instead of a cluster.
+ */
+public class TestHadoopFsReadPercent {
+
+ /**
+ * Operations per run of the tests that assert a split. Each one draws on its
+ * own, so a split is only exact at 0 and 100 percent; a run of this many
+ * leaves the share of reads far closer to the percentage than the tolerance
+ * of {@link #assertSplit} allows for.
+ */
+ private static final int OPS = 1000;
+
+ /** Operations of the runs that assert an exact count. */
+ private static final int FEW_OPS = 8;
+
+ @TempDir
+ private java.nio.file.Path tempDir;
+
+ private String rootPath;
+
+ @BeforeEach
+ void setUp() {
+ // toUri() rather than the path itself: on Windows the latter is not a valid
+ // URI, it has backslashes and a drive letter where the authority goes
+ String uri = tempDir.toUri().toString();
+ rootPath = uri.endsWith("/") ? uri.substring(0, uri.length() - 1) : uri;
+ CorruptingLocalFileSystem.corruptFromRead(Integer.MAX_VALUE);
+ }
+
+ /** Half of the operations read, unless the run asks for another split. */
+ @Test
+ void splitsOperationsEvenlyByDefault() {
+ CommandLine cmd = runValidator(1, OPS);
+
+ assertSplit(cmd, 50);
+ }
+
+ /**
+ * The requested share of the operations reads, whether that leaves the run
+ * write-heavy or read-heavy.
+ */
+ @ParameterizedTest
+ @ValueSource(ints = {10, 75, 90})
+ void splitsOperationsByPercent(int percent) {
+ CommandLine cmd =
+ runValidator(1, OPS, "--read-percent", String.valueOf(percent));
+
+ assertSplit(cmd, percent);
+ }
+
+ /** Nothing is read back at 0, which leaves a pure write load. */
+ @Test
+ void writesEveryOperationAtZeroPercent() {
+ CommandLine cmd = runValidator(1, FEW_OPS, "--read-percent", "0");
+
+ assertEquals(FEW_OPS, writeCount(cmd));
+ assertEquals(0, readCount(cmd));
+ }
+
+ /**
+ * Even at 100 a thread writes once: a read validates a file the thread wrote,
+ * and its first operation has nothing to read back yet.
+ */
+ @Test
+ void readsEveryOperationButTheFirstAtFullPercent() {
+ CommandLine cmd = runValidator(1, FEW_OPS, "--read-percent", "100");
+
+ assertEquals(1, writeCount(cmd));
+ assertEquals(FEW_OPS - 1, readCount(cmd));
+ }
+
+ /**
+ * Every read validates, not only the first or the last one: a run whose
+ * single write is read back three times fails wherever among the three the
+ * content is corrupted.
+ */
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2, 3})
+ void everyReadValidatesContent(int corruptedRead) {
+ CorruptingLocalFileSystem.corruptFromRead(corruptedRead);
+
+ int exitCode = new Freon().getCmd().execute(
+ "-D", "fs.file.impl=" + CorruptingLocalFileSystem.class.getName(),
+ "dfsrw",
+ "-r", rootPath,
+ "-p", "dfsrw-corrupt",
+ // the first operation writes, so the other three all read that file
+ "-n", "4",
+ "-t", "1",
+ "-s", "1KB",
+ "--buffer", "1024",
+ "--copy-buffer", "1024",
+ "--read-percent", "100");
+
+ assertNotEquals(0, exitCode,
+ "Corrupted content of read " + corruptedRead + " was not detected");
+ }
+
+ @Test
+ void rejectsPercentOutsideRange() {
+ assertThat(execute(new Freon().getCmd(), 1, FEW_OPS, "--read-percent", "-1"))
+ .isNotZero();
+ assertThat(execute(new Freon().getCmd(), 1, FEW_OPS, "--read-percent", "101"))
+ .isNotZero();
+ }
+
+ /**
+ * Every operation was a read or a write, and the reads are close enough to
+ * the requested share of them. The tolerance is six standard deviations of
+ * the draw: wide enough that a passing run is not chance, narrow enough to
+ * catch a split that ignores the percentage.
+ */
+ private static void assertSplit(CommandLine cmd, int percent) {
+ long reads = readCount(cmd);
+ assertEquals(OPS, reads + writeCount(cmd),
+ "every operation is either a read or a write");
+
+ double fraction = percent / 100.0;
+ long tolerance =
+ (long) Math.ceil(6 * Math.sqrt(OPS * fraction * (1 - fraction)));
+ assertThat(reads).isCloseTo(Math.round(OPS * fraction), within(tolerance));
+ }
+
+ private CommandLine runValidator(int threads, int ops, String... args) {
+ CommandLine cmd = new Freon().getCmd();
+ assertEquals(0, execute(cmd, threads, ops, args),
+ "Freon dfsrw command failed");
+ return cmd;
+ }
+
+ private int execute(CommandLine cmd, int threads, int ops, String... args) {
+ String[] fixed = {
+ "dfsrw",
+ "-r", rootPath,
+ "-p", "dfsrw",
+ "-n", String.valueOf(ops),
+ "-t", String.valueOf(threads),
+ "-s", "1KB",
+ "--buffer", "1024",
+ "--copy-buffer", "1024"};
+ String[] argv = new String[fixed.length + args.length];
+ System.arraycopy(fixed, 0, argv, 0, fixed.length);
+ System.arraycopy(args, 0, argv, fixed.length, args.length);
+ return cmd.execute(argv);
+ }
+
+ private static long writeCount(CommandLine cmd) {
+ return timerCount(cmd, "file-write");
+ }
+
+ private static long readCount(CommandLine cmd) {
+ return timerCount(cmd, "file-read-validate");
+ }
+
+ private static long timerCount(CommandLine cmd, String name) {
+ BaseFreonGenerator subject = (BaseFreonGenerator)
+ cmd.getParseResult().subcommand().commandSpec().userObject();
+ return subject.getMetrics().timer(name).getCount();
+ }
+
+ /**
+ * A {@link LocalFileSystem} that alters the content it hands out from the
+ * n-th {@code open()} on. The bytes are changed above the checksum
+ * verification of {@link org.apache.hadoop.fs.ChecksumFileSystem}, so it is
+ * the workload, not the file system, that has to notice.
+ */
+ public static final class CorruptingLocalFileSystem extends LocalFileSystem {
+
+ private static final AtomicInteger READS = new AtomicInteger();
+ private static int corruptFrom = Integer.MAX_VALUE;
+
+ static void corruptFromRead(int read) {
+ READS.set(0);
+ corruptFrom = read;
+ }
+
+ @Override
+ public FSDataInputStream open(Path f, int bufferSize) throws IOException {
+ FSDataInputStream input = super.open(f, bufferSize);
+ // the hidden .crc companions are the checksum bookkeeping of the file
+ // system itself, only the reads of the workload are to be counted
+ if (f.getName().startsWith(".") || READS.incrementAndGet() < corruptFrom) {
+ return input;
+ }
+ return new FSDataInputStream(new FlippingInputStream(input));
+ }
+ }
+
+ /**
+ * Passes the wrapped stream through, with the first byte of the file flipped.
+ * {@link FSInputStream} supplies the positioned reads on top of seek and
+ * read, so only those two and the plain reads are forwarded here.
+ */
+ private static final class FlippingInputStream extends FSInputStream {
+
+ private final FSDataInputStream input;
+
+ private FlippingInputStream(FSDataInputStream input) {
+ this.input = input;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ long pos = input.getPos();
+ int read = input.read(b, off, len);
+ if (read > 0 && pos == 0) {
+ b[off] ^= 0xff;
+ }
+ return read;
+ }
+
+ @Override
+ public int read() throws IOException {
+ long pos = input.getPos();
+ int b = input.read();
+ return b >= 0 && pos == 0 ? b ^ 0xff : b;
+ }
+
+ @Override
+ public void close() throws IOException {
+ input.close();
+ }
+
+ @Override
+ public void seek(long pos) throws IOException {
+ input.seek(pos);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return input.getPos();
+ }
+
+ @Override
+ public boolean seekToNewSource(long targetPos) throws IOException {
+ return input.seekToNewSource(targetPos);
+ }
+ }
+}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopFsReadWriteValidator.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopFsReadWriteValidator.java
index 9bff0f416f27..236516170671 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopFsReadWriteValidator.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopFsReadWriteValidator.java
@@ -85,7 +85,8 @@ public void testWriteReadValidate(BucketLayout layout) throws Exception {
String rootPath = OZONE_URI_SCHEME + "://" + bucketName + "." + volumeName;
String om = cluster().getConf().get(OZONE_OM_ADDRESS_KEY);
- int exitCode = new Freon().getCmd().execute(
+ CommandLine cmd = new Freon().getCmd();
+ int exitCode = cmd.execute(
"-D", OZONE_OM_ADDRESS_KEY + "=" + om,
"dfsrw",
"-n", String.valueOf(fileCount),
@@ -96,12 +97,20 @@ public void testWriteReadValidate(BucketLayout layout) throws Exception {
);
assertEquals(0, exitCode, "Freon dfsrw command failed");
+ // -n counts operations, and --read-percent decides per operation whether it
+ // writes or reads one back, so how many of them wrote is drawn rather than
+ // fixed. Each write takes a path of its own here, so the run leaves
+ // somewhere between one file and one per operation.
+ BaseFreonGenerator subject = (BaseFreonGenerator)
+ cmd.getParseResult().subcommand().commandSpec().userObject();
+ assertEquals(fileCount, subject.getSuccessCount());
+
// verify all files were written with the requested size
OzoneConfiguration conf = new OzoneConfiguration(cluster().getConf());
try (FileSystem fileSystem = FileSystem.get(URI.create(rootPath), conf)) {
FileStatus[] files =
fileSystem.listStatus(new Path(rootPath + "/" + prefix));
- assertEquals(fileCount, files.length, "Unexpected number of files");
+ assertThat(files.length).isBetween(1, fileCount);
Set