taklwu commented on code in PR #11783:
URL: https://github.com/apache/ozone/pull/11783#discussion_r4393344421
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java:
##########
@@ -327,15 +335,56 @@ 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]);
Review Comment:
or at least have another checkOpen here
```suggestion
checkOpen();
locations.put(index, dataLocations[index]);
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java:
##########
@@ -327,15 +335,56 @@ 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);
+ // A concurrent close fails the reads, although the locations are not bad,
so they must not be reported.
+ checkOpen();
+ if (unexpected != null) {
+ Throwables.throwIfUnchecked(unexpected);
+ throw new IOException(unexpected);
+ }
+ BadDataLocationException failure = readFailure(results, locations::get);
+ if (failure != null) {
Review Comment:
nit: or if the stream closed, the failure could be different.
```suggestion
if (failure != null) {
checkOpen();
```
--
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]
