smengcl commented on code in PR #11783:
URL: https://github.com/apache/ozone/pull/11783#discussion_r4385162611
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java:
##########
@@ -327,15 +335,54 @@ protected long remaining() {
return blockLength() - position;
}
+ /**
+ * Reads the cells covered by the buffer from the internal block streams of
their data indexes with positioned
+ * reads, so neither this stream nor its internal streams change position,
and calls may run concurrently with each
+ * other and with sequential reads. The indexes are read in parallel if the
parallel read executor is set.
+ *
+ * <p>A failed index is not switched to a spare. Instead, a {@link
BadDataLocationException} carrying every failed
+ * index is thrown, so that the caller can fall back to a reconstruction
read for this call.
+ */
@Override
- protected int readPositioned(long offset, ByteBuffer buffer) {
- throw new UnsupportedOperationException("Use ECBlockInputStreamProxy for
positioned reads");
- }
+ protected int readPositioned(long offset, ByteBuffer buffer) throws
IOException {
+ checkOpen();
+ final long available = blockLength() - offset;
+ if (available <= 0) {
+ return EOF;
+ }
+ final int length = (int) Math.min(buffer.remaining(), available);
+ final int bufferStart = buffer.position();
- @Override
- public boolean hasCapability(String capability) {
- return !StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)
- && super.hasCapability(capability);
+ final Map<Integer, DatanodeDetails> locations = new TreeMap<>();
+ final List<FutureTask<IndexReadResult>> tasks = new ArrayList<>();
+ for (Map.Entry<Integer, List<CellRead>> entry : splitCells(offset, length,
new ArrayList<>()).entrySet()) {
+ int index = entry.getKey();
+ BlockExtendedInputStream stream;
+ synchronized (streamLock) {
+ locations.put(index, dataLocations[index]);
+ try {
+ stream = getOrOpenStream(index);
+ } catch (IOException e) {
+ throw new BadDataLocationException(dataLocations[index], index, e);
+ }
+ }
+ List<CellRead> indexCells = entry.getValue();
+ tasks.add(new FutureTask<>(() -> readCells(stream, indexCells, buffer,
bufferStart, false, true)));
+ }
+
+ Map<Integer, IndexReadResult> results = new TreeMap<>();
+ Throwable unexpected = runIndexReads(tasks, results);
+ if (unexpected != null) {
+ Throwables.throwIfUnchecked(unexpected);
+ throw new IOException(unexpected);
+ }
+ BadDataLocationException failure = readFailure(results, locations::get);
+ if (failure != null) {
+ throw failure;
Review Comment:
When this reader is closed while a positioned read is in flight, the read
reports healthy datanodes as bad. The common trigger is a sequential read on
the same `ECBlockInputStreamProxy` failing over: `failoverToReconstructionRead`
calls `blockReader.close()`, which closes every per-index `BlockInputStream`. A
concurrent pread that already holds those streams then gets
`IOException("BlockInputStream has been closed.")` from
`BlockInputStream.readPositioned`, `readCells` records it as an index error,
and this line throws a `BadDataLocationException` carrying the locations of
those indexes. The proxy adds them to the shared `failedLocations`, so the
fallback reader and every later pread of this block exclude them. For example,
with RS(3,2), one dead datanode plus a 1 MB pread touching the other two data
indexes leaves 3 failed locations, and every later pread of the block fails
with `InsufficientLocationsException` although the data is readable. Before
this change each pread used its
own reader, so a sequential failover could not affect it.
Checking whether this stream was closed before blaming the locations avoids
that (setting `closed` before `closeStreams()` closes the window where the
streams are closed but the flag is not yet set):
```diff
BadDataLocationException failure = readFailure(results, locations::get);
if (failure != null) {
+ // a concurrent close fails the reads without the locations being bad
+ checkOpen();
throw failure;
}
@@ public synchronized void close() {
LOG.debug("{}: close", this);
- closeStreams();
closed = true;
+ closeStreams();
}
```
Relatedly, if the closed stream had not been initialized yet (for example
one this pread just opened), `BlockInputStream.readPositioned` calls
`initialize()` on it and throws `NullPointerException` from
`ContainerProtocolCalls.getBlock` because the client is null. That is rethrown
by `Throwables.throwIfUnchecked` above and escapes the proxy, which only falls
back on `IOException`. Calling `checkOpen()` at the start of
`BlockInputStream.initialize()` (after the `initialized` check) would turn it
into an `IOException` as well.
--
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]
