smengcl commented on code in PR #11501: URL: https://github.com/apache/ozone/pull/11501#discussion_r4300133664
########## hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/TestS3SDKWithEcParallelRead.java: ########## @@ -0,0 +1,34 @@ +/* + * 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.s3.awssdk; + +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.scm.ScmConfigKeys; +import org.apache.hadoop.ozone.OzoneConfigKeys; + +/** Runs the S3 SDK tests with S3 Gateways that read the cells of EC keys in parallel. */ +class TestS3SDKWithEcParallelRead extends OzoneS3SDKTests { Review Comment: This starts another cluster and reruns the whole S3 SDK suite to flip one client flag, but only the multi-cell EC GET exercises the new path. Could the parallel case run on the existing cluster instead, e.g. as a parameterized variant of the EC GET test added to `AbstractS3SDKV2Tests`? That would avoid roughly doubling the S3 SDK test time. ########## hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestEcParallelReadBenchmark.java: ########## @@ -0,0 +1,442 @@ +/* + * 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.client.rpc.read; + +import static org.apache.hadoop.hdds.client.ECReplicationConfig.EcCodec.RS; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Random; +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.zip.CRC32; +import org.apache.hadoop.hdds.client.ECReplicationConfig; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.apache.hadoop.hdds.utils.IOUtils; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.client.ObjectStore; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.OzoneClientFactory; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Benchmark for reading the cells of a healthy EC block in parallel (HDDS-16867). + * + * <p>Every EC key is read through two sets of clients of the same cluster: with + * {@code ozone.client.ec.parallel.read.enabled=false} (cells are read one at a time) and with it set to true. + * Each client has its own datanode connections. Rounds alternate the order of the two arms, after a warm-up, and + * the median throughput of the rounds is reported; latency percentiles are taken over all measured rounds. + * + * <p>The benchmark runs on a 5 and on a 9 datanode cluster. rs-6-3 needs 9 datanodes, so on 5 datanodes only + * rs-3-2 is measured; there every datanode holds a cell of every key, while on 9 datanodes the cells of rs-3-2 + * keys are spread over more datanodes. + * + * <p>All datanodes run in this JVM and are reached over loopback, so a datanode round trip costs far less than + * on a real network. The speed-up measured here comes from overlapping the datanode read, the transfer and the + * client-side checksum verification of different cells; on a real network the serial round trips that the + * parallel path removes cost more, so this is a lower bound. With several readers, the readers and all the + * datanodes share the CPUs of one machine. + * + * <p>Run with: + * <pre> + * mvn -pl :ozone-integration-test test -Dtest=TestEcParallelReadBenchmark \ + * -Dexcluded-test-groups= -DskipShade -DskipRecon + * </pre> + */ +@Tag("benchmark") +public class TestEcParallelReadBenchmark { Review Comment: Since this is tagged `benchmark` and never runs in CI, it is likely to rot. Would it be better to leave it out of the tree (or move it to a separate JIRA) and keep the results in the PR description? Similarly, moving the audit log files under `target/` looks useful but is independent of this feature, so it could be its own small change. ########## hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java: ########## @@ -391,6 +421,189 @@ protected synchronized int readWithStrategy(ByteReaderStrategy strategy) return totalRead; } + /** + * Returns true if the parallel read path is enabled and the next read into the buffer covers more than one cell. + */ + private boolean spansMultipleCells(ByteBuffer buffer) { + if (parallelReadExecutor == null || closed) { + return false; + } + long length = Math.min(buffer.remaining(), remaining()); + return position % ecChunkSize + length > ecChunkSize; + } + + /** + * Reads the cells covered by the buffer from the internal block streams of their data indexes in parallel. The + * cells of one data index, which come from consecutive stripes, are read in order by a single task, so the + * fan-out is at most the number of data units. The calling thread runs the task of the first cell, and any task + * the executor has not started yet, and returns only after all tasks have finished. + * + * <p>Like {@link #readWithStrategy}, the position and the buffer are advanced past the data that was read. If a + * read fails, they are advanced only by the cells before the first failed cell, and a + * {@link BadDataLocationException} carrying every failed index is thrown. + */ + private int readCellsInParallel(ByteBuffer buffer) throws IOException { + final long startPosition = position; + final int length = (int) Math.min(buffer.remaining(), remaining()); + final int bufferStart = buffer.position(); + + // Split the range into cells, grouped by data index in the order they are first needed. + final List<CellRead> cells = new ArrayList<>(); + final Map<Integer, List<CellRead>> cellsByIndex = new LinkedHashMap<>(); + for (int offset = 0; offset < length;) { + long cellPosition = startPosition + offset; + int cellLength = (int) Math.min(ecChunkSize - cellPosition % ecChunkSize, length - offset); + int index = (int) ((cellPosition / ecChunkSize) % repConfig.getData()); + List<CellRead> indexCells = cellsByIndex.computeIfAbsent(index, k -> new ArrayList<>()); + CellRead cell = new CellRead(index, indexCells.size(), cellPosition, offset, cellLength); + indexCells.add(cell); + cells.add(cell); + offset += cellLength; + } + + // getOrOpenStream is not thread safe, so open every stream before reading. + final List<FutureTask<IndexReadResult>> tasks = new ArrayList<>(); + for (Map.Entry<Integer, List<CellRead>> entry : cellsByIndex.entrySet()) { + int index = entry.getKey(); + BlockExtendedInputStream stream; + try { + stream = getOrOpenStream(index); + } catch (IOException e) { + throw new BadDataLocationException(dataLocations[index], index, e); + } + List<CellRead> indexCells = entry.getValue(); + boolean checkPosition = seeked; + tasks.add(new FutureTask<>(() -> readCells(stream, indexCells, buffer, bufferStart, checkPosition))); + } + + ExecutorService executor = parallelReadExecutor.get(); + for (int i = 1; i < tasks.size(); i++) { + try { + executor.execute(tasks.get(i)); + } catch (RejectedExecutionException e) { + // the task is run by this thread below + LOG.debug("{}: parallel read task rejected", this, e); + } + } + // FutureTask.run does nothing if the task has already been started by the executor. + for (FutureTask<IndexReadResult> task : tasks) { + task.run(); + } + // Wait for every task, even after an unexpected error, as they all write into the caller's buffer. + Map<Integer, IndexReadResult> results = new TreeMap<>(); + Throwable unexpected = null; + for (FutureTask<IndexReadResult> task : tasks) { + try { + IndexReadResult result = Uninterruptibles.getUninterruptibly(task); + results.put(result.index, result); + } catch (ExecutionException e) { + unexpected = unexpected == null ? e.getCause() : unexpected; + } + } + if (unexpected != null) { + Throwables.throwIfUnchecked(unexpected); + throw new IOException(unexpected); + } + + // A failure on one index must not be hidden by the success of a later cell, so only advance by the cells + // before the first failed cell. + int read = 0; + for (CellRead cell : cells) { + if (cell.ordinal >= results.get(cell.index).cellsRead) { + break; + } + read += cell.length; + } + position = startPosition + read; Review Comment: Minor: when this method ends by throwing, `position` only moves past the cells before the first failure (and not at all when line 504 rethrows an unchecked exception). Indexes whose tasks succeeded have already moved their inner streams further, though. If `seeked` is still false, a later read on this stream skips the position check in `readCells` and `seekStreamIfNecessary` and reads from the wrong offset. The proxy closes this reader on `BadDataLocationException`, so the main path is safe, but a caller that retries after a `RuntimeException` would silently get wrong data. Setting `seeked = true` before throwing would cover it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
