yandrey321 commented on code in PR #11501:
URL: https://github.com/apache/ozone/pull/11501#discussion_r4304015578
##########
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:
Good catch, fixed.
After all tasks finish, readCellsInParallel now sets seeked = true if any
task failed (IOException or unchecked), before either throw, so the next read
checks and re-seeks every inner stream. Added
TestECBlockInputStream#testReadAfterFailedParallelReadReturnsCorrectData: index
2 fails on a 3-cell read, the error is cleared, and a retry on the same stream
without a seek must return the right bytes; without the fix it returns index
3's data from the wrong offset. The serial readWithStrategy path has a narrower
version of this (a cell failing mid-read can leave that one stream ahead of
position); that predates this change, so I left it for a follow-up JIRA
--
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]
