smengcl commented on code in PR #11501:
URL: https://github.com/apache/ozone/pull/11501#discussion_r4268083979
##########
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));
Review Comment:
These tasks also run `BlockInputStream`'s pipeline/token refresh, but secure
S3 request authentication is stored in a plain `ThreadLocal` on the request
thread. If an index handled by a worker needs to refresh after
`CONTAINER_NOT_FOUND` or a connectivity error, the callback reaches
`OzoneManagerProtocolClientSideTranslatorPB.submitRequest()` without that
authentication and throws `IllegalArgumentException: S3 Auth expected to be set
but is null`. The unchecked rethrow below bypasses spare-location retry and the
proxy's reconstruction fallback, so a recoverable EC GET fails. This is
reachable with the gateway's default 4 MB copy buffer and 1 MB EC cells when
the new flag is enabled.
Please preserve the request authentication during worker-side metadata
refresh and restore the worker's previous context afterward. A regression test
should trigger an OM refresh from a worker with the S3 auth check enabled; the
current serial path performs that refresh on the authenticated request thread.
--
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]
