yandrey321 commented on code in PR #11348:
URL: https://github.com/apache/ozone/pull/11348#discussion_r4188486706
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java:
##########
@@ -195,45 +180,35 @@ public synchronized void seek(long pos) throws
IOException {
}
@Override
- public boolean readFully(long position, ByteBuffer buffer) throws
IOException {
- if (!statelessPositionedReadSupported) {
- return false;
- }
- return readFullyStateless(position, buffer);
- }
-
- /**
- * Shared stateless positioned-read loop across parts using {@link
PartInputStream#readPositioned}.
- */
- private boolean readFullyStateless(long position, ByteBuffer buffer) throws
IOException {
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ checkOpen();
if (!buffer.hasRemaining()) {
- return true;
+ return 0;
}
-
long pos = position;
int bytesRead = 0;
- while (buffer.hasRemaining()) {
- if (pos < 0 || pos >= length) {
- if (bytesRead > 0) {
- return true;
- }
- throw new EOFException("EOF encountered at pos: " + pos +
- " for key: " + key);
+ while (buffer.hasRemaining() && pos < length) {
Review Comment:
Two problems, both from this method being unsynchronized while the state it
reads is
mutable.
(a) Visibility. `length` (line 37) is neither final nor volatile. It is
written in the
constructor and again in `synchronized initialize()` (line 242). A
positioned read
racing a first sequential op can see either value with no happens-before
edge. Making
`length` volatile is the minimal fix and costs nothing on this path.
(`closed` was
correctly made volatile in this patch — `length` needs the same treatment.)
(b) The clamp can go negative. `partOffsets` is computed once in the
constructor from
pre-initialize part lengths and is never recomputed, while `initialize()`
recomputes
`length` and, for under-construction blocks,
BlockInputStream.initialize():167 resets
`length = blockData.getSize()` from what the datanode actually has. If a
part's length
comes back *shorter* than the OM-reported length the offsets were built
from, then for
a `pos` that is still < the new `length`, `part.getLength() - partPosition`
is
negative and `buffer.limit(buffer.position() + negative)` throws
IllegalArgumentException — a RuntimeException escaping a read path that
callers only
guard for IOException. (The offset staleness itself predates this PR; the
sequential
path at lines 149-155/178 shares it. What is new is that the clamp converts
it into an
unchecked throw.)
Both are cheap to close:
final long available = part.getLength() - partPosition;
if (available <= 0) {
break; // offsets and lengths disagree; treat as EOF
}
buffer.limit(buffer.position() + (int) Math.min(buffer.remaining(),
available));
and recomputing `partOffsets` in initialize() (or documenting on the field
that it is
only valid for a fully-committed key) would make the invariant explicit
rather than
implicit.
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/CapableOzoneFSInputStream.java:
##########
@@ -34,8 +35,9 @@ public boolean hasCapability(String capability) {
switch (StringUtils.toLowerCase(capability)) {
case StreamCapabilities.READBYTEBUFFER:
case StreamCapabilities.UNBUFFER:
- case StreamCapabilities.PREADBYTEBUFFER:
return true;
+ case StreamCapabilities.PREADBYTEBUFFER:
Review Comment:
This check is a tautology in production. The wrapped stream here is always an
OzoneInputStream (built in RpcClient.createInputStream and handed to
CapableOzoneFSInputStream by OzoneFileSystem:114 /
RootedOzoneFileSystem:112), and
this PR makes OzoneInputStream extend ExtendedInputStream, which *implements*
ByteBufferPositionedReadable unconditionally. So the answer is `true` for
every key,
including a GDPR key, where RpcClient:2682 wraps a
javax.crypto.CipherInputStream and
OzoneInputStream.readPositioned:83 throws UnsupportedOperationException.
A caller that does the right thing — probe hasCapability(PREADBYTEBUFFER),
then call
read(long, ByteBuffer) — gets told yes and then gets a RuntimeException.
OzoneInputStream.hasCapability already computes this correctly (it tests its
own
delegate). Delegate to it instead of re-deriving from the type:
case StreamCapabilities.PREADBYTEBUFFER:
return getWrappedInputStream() instanceof StreamCapabilities
&& ((StreamCapabilities)
getWrappedInputStream()).hasCapability(capability);
Note the new unit test does not catch this because it bypasses the production
wrapping: cursorOnlyStreamsDoNotEmulatePositionedReads constructs
`new CapableOzoneFSInputStream(new SeekableOnlyInputStream(source), null)` —
a raw
delegate rather than an OzoneInputStream — so the instanceof legitimately
answers
false there. Please add a case that wraps the delegate in OzoneInputStream,
which is
the only shape the filesystem ever sees.
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java:
##########
@@ -66,6 +70,38 @@ public int read(ByteBuffer byteBuffer) throws IOException {
}
}
+ @Override
+ protected int readWithStrategy(ByteReaderStrategy strategy) throws
IOException {
+ return strategy.readFromBlock(inputStream, strategy.getTargetLength());
+ }
+
+ @Override
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ if (inputStream instanceof ByteBufferPositionedReadable) {
+ return ((ByteBufferPositionedReadable) inputStream).read(position,
buffer);
+ }
+ throw new UnsupportedOperationException("Positioned reads are not
supported by "
+ + inputStream.getClass().getName());
+ }
+
+ @Override
+ public boolean hasCapability(String capability) {
+ if (!super.hasCapability(capability)) {
+ return false;
+ }
+ if (inputStream instanceof StreamCapabilities
+ && !((StreamCapabilities) inputStream).hasCapability(capability)) {
+ return false;
+ }
+ if (StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)) {
+ return inputStream instanceof ByteBufferPositionedReadable;
+ }
+ if (StreamCapabilities.READBYTEBUFFER.equalsIgnoreCase(capability)) {
+ return inputStream instanceof ByteBufferReadable;
+ }
+ return inputStream instanceof CanUnbuffer;
Review Comment:
This is the fall-through for "whatever capability the base class said true
to that
isn't one of the two above" — correct only because the base switch has
exactly three
cases. If ExtendedInputStream ever adds a fourth, this silently answers the
wrong
question. Make it explicit:
if (StreamCapabilities.UNBUFFER.equalsIgnoreCase(capability)) {
return inputStream instanceof CanUnbuffer;
}
return false;
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java:
##########
@@ -66,6 +70,38 @@ public int read(ByteBuffer byteBuffer) throws IOException {
}
}
+ @Override
+ protected int readWithStrategy(ByteReaderStrategy strategy) throws
IOException {
Review Comment:
This is dead code: read(), read(byte[],int,int) and read(ByteBuffer) at
:53-:71 all
delegate straight to `inputStream` and never reach the base class's
`read(ByteReaderStrategy)`. Two consequences worth deciding on deliberately:
- the base class's sequential entry points are `synchronized`, these
overrides are
not, so OzoneInputStream's sequential reads have different locking from
every
other ExtendedInputStream;
- if the strategy path is genuinely unreachable, implementing it invites a
future
reader to assume it is live.
Either drop the direct overrides and let the strategy path do the work
(uniform
locking, one code path), or keep them and make readWithStrategy throw
UnsupportedOperationException with a comment saying why.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java:
##########
@@ -195,45 +180,35 @@ public synchronized void seek(long pos) throws
IOException {
}
@Override
- public boolean readFully(long position, ByteBuffer buffer) throws
IOException {
- if (!statelessPositionedReadSupported) {
- return false;
- }
- return readFullyStateless(position, buffer);
- }
-
- /**
- * Shared stateless positioned-read loop across parts using {@link
PartInputStream#readPositioned}.
- */
- private boolean readFullyStateless(long position, ByteBuffer buffer) throws
IOException {
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ checkOpen();
if (!buffer.hasRemaining()) {
- return true;
+ return 0;
}
-
long pos = position;
int bytesRead = 0;
- while (buffer.hasRemaining()) {
- if (pos < 0 || pos >= length) {
- if (bytesRead > 0) {
- return true;
- }
- throw new EOFException("EOF encountered at pos: " + pos +
- " for key: " + key);
+ while (buffer.hasRemaining() && pos < length) {
+ int index = binarySearchOffsetIndex(partOffsets, pos);
+ PartInputStream part = partStreams.get(index);
+ long partPosition = pos - partOffsets[index];
+ final int oldLimit = buffer.limit();
+ final int n;
+ try {
+ buffer.limit(buffer.position() + (int) Math.min(buffer.remaining(),
part.getLength() - partPosition));
+ n = part.read(partPosition, buffer);
+ } finally {
+ buffer.limit(oldLimit);
+ }
+ if (n < 0) {
+ break;
}
- int idx = binarySearchOffsetIndex(partOffsets, pos);
- long partPos = pos - partOffsets[idx];
- int n = partStreams.get(idx).readPositioned(partPos, buffer);
- if (n <= 0) {
- if (bytesRead > 0) {
- return true;
- }
- throw new EOFException("EOF encountered at pos: " + pos +
- " for key: " + key);
+ if (n == 0) {
Review Comment:
binarySearchOffsetIndex relies on Arrays.binarySearch, whose result among
*duplicate*
offsets is unspecified. A zero-length part produces a duplicate offset, so
the search
can land on it; the clamp at line 197 then sets the limit to the current
position,
part.read() correctly returns 0 for an empty buffer, and this throws — for a
stream
that is in no way broken. The method created the empty buffer itself, so 0
here is not
a protocol violation.
`continue` (after advancing past the part) rather than throwing would be
correct. Same
`available <= 0` guard suggested above covers it.
I could not construct a zero-length part from today's write paths, so this is
robustness rather than a live bug — but the throw is reachable from a state
the method
itself produces, which is worth closing cheaply.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java:
##########
@@ -197,6 +197,23 @@ public synchronized int read(ByteBuffer buf) throws
IOException {
return totalRead;
}
+ @Override
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ ensureNotClosed();
+ if (!buffer.hasRemaining()) {
+ return 0;
+ }
+ if (position >= blockInfo.getLength()) {
+ return EOF;
+ }
+ // Each request owns its cursor, reconstruction buffers and failure
recovery.
+ try (ECBlockInputStreamProxy reader = new
ECBlockInputStreamProxy(repConfig, blockInfo,
Review Comment:
Correctness is fine and the new test proves readers are closed and reads
overlap. The
concern is what this costs per call. Each positioned read constructs and
tears down a
complete EC reader: on the direct path, data-count BlockInputStreams with
their own
xceiver clients; on the reconstruction path, a stripe reader holding
ByteBufferPool
buffers. read(ByteBuffer) does loop internally, so it is one reader per
block per
pread rather than per chunk — but for a 4KB pread against a 3-2 EC key that
is still
three fresh datanode streams built and discarded.
More importantly, the fresh proxy starts with an empty `failedLocations`. A
replica
that the outer stream has already failed over on (BadDataLocationException)
is
re-attempted by every subsequent positioned read, so each one pays the
connect/read
timeout to the dead datanode again before falling back to reconstruction.
Under a
positioned-read workload against a key with one bad replica, that is a
per-call
penalty that never amortizes.
Worth at least a comment acknowledging it, and ideally passing the outer
`failedLocations` into the per-request reader (the ECBlockInputStreamFactory
signature
already takes a failed-locations list). If sharing the set across threads is
the
obstacle, a CopyOnWriteArrayList or a snapshot-plus-merge would do.
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java:
##########
@@ -193,13 +188,23 @@ public int read(long position, byte[] array, int offset,
int length) throws IOEx
return readImpl(position, ByteBuffer.wrap(array, offset, length));
}
+ protected InputStream getWrappedInputStream() {
+ return inputStream;
+ }
+
private int readImpl(long position, ByteBuffer buffer) throws IOException {
Review Comment:
Both byte-array overloads now route here — read(long, byte[], int, int):183
and
readFully(long, byte[], int, int):228 — so they throw for any delegate that
cannot do
ByteBuffer positioned reads. That is a functional regression, not just a
capability
question: read(long, byte[], int, int) and readFully are plain
PositionedReadable
methods. Hadoop callers invoke them with no capability probe at all
(FSDataInputStream
exposes them directly; ORC, Parquet and Hadoop's own IOUtils.readFully use
them), and
there is no StreamCapabilities constant that gates them.
Before this patch, bestEffortRead/bestEffortReadFullySynchronized served
these under
positionedReadLock via seek-read-restore. The javadoc you deleted named
exactly the
case that now breaks:
"Uses read(ByteBuffer) -- which handles streams that do not implement
ByteBufferReadable (e.g. GDPR javax.crypto.CipherInputStream)"
Reaching concurrency by deleting the only path that worked for a whole class
of keys
is the wrong trade. Options, in order of preference:
a) Keep the locked seek-read-restore fallback for the byte-array overloads
only,
reached when the delegate is not ByteBufferPositionedReadable.
Concurrency is
unchanged for every stream that *does* support native pread; GDPR keys
keep
working, serialized, as they do today.
b) Make the GDPR path capable at the source — RpcClient:2682 could wrap the
KeyInputStream in something that implements
ByteBufferPositionedReadable with
per-call Cipher instances — which is a larger change and probably its
own Jira.
Either way the PR description should say that GDPR-encrypted keys are
affected; right
now it reads as purely additive.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java:
##########
@@ -100,6 +128,7 @@ public synchronized boolean seekToNewSource(long l) throws
IOException {
@Override
public boolean hasCapability(String capability) {
Review Comment:
Advertising PREADBYTEBUFFER in the base class means every present and future
ExtendedInputStream claims it by default, and a subclass that cannot honour
it has to
remember to say so. Two already have to, and the resulting override is hard
to read:
// ECBlockInputStream:xxx and ECBlockReconstructedInputStream:xxx
return !StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)
&& super.hasCapability(capability);
The previous shape (`readFully(long, ByteBuffer)` returning false until
overridden)
was opt-in, which fails safe. Given that finding #1 above is exactly a
false-positive capability report, I would rather see this as an overridable
hook —
e.g. a `protected boolean supportsPositionedRead() { return true; }` that
the two EC
classes return false from, with the switch consulting it — than as a default
that two
classes already have to undo.
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java:
##########
@@ -66,6 +70,38 @@ public int read(ByteBuffer byteBuffer) throws IOException {
}
}
+ @Override
+ protected int readWithStrategy(ByteReaderStrategy strategy) throws
IOException {
+ return strategy.readFromBlock(inputStream, strategy.getTargetLength());
+ }
+
+ @Override
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
Review Comment:
The no-arg constructor at :40 leaves `inputStream` null, so this NPEs on
`getClass()` instead of throwing the intended UnsupportedOperationException.
A
null-guard in the message (or in the constructor) is a one-liner.
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java:
##########
@@ -20,15 +20,19 @@
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
import org.apache.hadoop.fs.ByteBufferReadable;
import org.apache.hadoop.fs.CanUnbuffer;
import org.apache.hadoop.fs.Seekable;
+import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
+import org.apache.hadoop.hdds.scm.storage.ExtendedInputStream;
/**
* OzoneInputStream is used to read data from Ozone.
* It uses {@link KeyInputStream} for reading the data.
*/
-public class OzoneInputStream extends InputStream implements CanUnbuffer,
+public class OzoneInputStream extends ExtendedInputStream implements
CanUnbuffer,
Review Comment:
- The `implements` clause is now entirely redundant — ExtendedInputStream
already
implements all three.
- This changes the superclass of a public client API type (returned by
OzoneBucket.readKey()) from java.io.InputStream to
org.apache.hadoop.fs.FSInputStream. It stays binary- and source-compatible
(InputStream remains an ancestor, and the added members are all new), but
it does
put hadoop-common's fs hierarchy into the public client type and adds a
dozen
public methods to it. Worth a line in the PR description so reviewers can
weigh it
rather than discover it.
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java:
##########
Review Comment:
ExtendedInputStream:45 checks these in the opposite order, so `read(-1,
emptyBuffer)`
returns 0 here and throws EOFException there. Likewise read(long, byte[],
int, int):184
returns 0 for `length == 0` *before* validatePositionedReadArgs, which skips
the
null-buffer and negative-position checks that Hadoop's contract specifies.
Low stakes, but the two layers should agree, and validating first is the
cheaper
convention to defend. The new test only exercises the negative position with
a
non-empty buffer, so the divergence is currently untested.
--
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]
