yandrey321 commented on code in PR #11501:
URL: https://github.com/apache/ozone/pull/11501#discussion_r4283447635
##########
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:
ood catch. Fixed in 9fd41ba50b, and I extended it to the read-consistency
hint, which had the same problem.
- **Fix:** the parallel-read pool's `execute` wraps every task. The wrapper
takes the submitter's values of the OM
client's request-context thread-locals (S3 auth and the read-consistency
hint) and sets them on the worker for the
duration of the task. Afterwards it restores the worker's previous values,
or removes them if the worker had
none, including when the task throws. Pool threads therefore never keep a
request's credentials after the task
ends. The thread-locals are taken from `OzoneManagerClientProtocol`
(`getS3CredentialsProvider()` and the new
`getReadConsistencyHintProvider()`), so the list sits in
`RpcClient.requestContext()`.
- **Regression test:**
`TestOzoneECClient.testLocationRefreshDuringReadKeepsRequestContext` runs in
both read modes.
It sets S3 auth and a `LOCAL_LEASE` hint on the reading thread, then fails
one GetBlock with
`CONTAINER_NOT_FOUND` so that a pool thread refreshes the block locations
from OM. It asserts that the failure was
served on an `ec-parallel-reader-` thread and that every OM request
carries both. Without the fix it fails in
parallel mode only, with `GetKeyInfo sent without S3 authentication` and
`expected LOCAL_LEASE but was
READ_CONSISTENCY_UNSPECIFIED`.
`TestRpcClient.testWithSubmitterContextPropagatesAndRestores` covers the
propagate/restore/clear cases of the wrapper.
- **End to end:** `TestSecureOzoneRpcClient` and the S3 SDK suites now read
multi-cell EC keys under S3 auth and a
read-consistency hint in both modes. `TestS3SDKWithEcParallelRead` runs
the S3 SDK suite with parallel-read
gateways. These cannot force a worker-side refresh, so the unit test above
is the one that catches this bug.
--
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]
