HDFS-3246: pRead equivalent for direct read path #597
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
💔 -1 overall
This message was automatically generated. |
|
Rebased patch, fixed remaining cc issues. Not sure why all those tests failed, they all pass locally. Let's see if Hadoop QA hits them again. |
|
💔 -1 overall
This message was automatically generated. |
openinx
reviewed
Mar 21, 2019
| } | ||
|
|
||
| return n; | ||
| } catch (ClassCastException e) { |
There was a problem hiding this comment.
Here we catch the ClassCastException to check whether the inpustream has implemented an ByteBufferPositionedReadable interface ? I don't think it's a good way, because the catch{...} block need an full stack and not so efficient .
How about use the following :
if(in instanceof ByteBufferPositionedReadable){
//...
}else{
throw new UnsupportedOperationException(...)
}There was a problem hiding this comment.
Yeah, its probably not the most efficient way to do things, but thats how all other methods handle the same thing: PositionedReadable, Seekable, etc. + the exception handling code wouldn't be on the hot path. So in this case I would prefer to keep the code consistent with the rest of the class.
There was a problem hiding this comment.
hrm. I think we should probably do a follow-up JIRA to fix this, not for performance reasons, but because the try{...} block encompasses a lot of code. Let's say we accidentally screw up something in our encryption config and we get a ClassCastException somewhere inside decrypt. We'll swallow the real exception and claim that positioned read isn't supported, which isn't quite right.
So, I agree an instanceof check up front is probably the clearest from a code perspective and also avoids the above issue.
There was a problem hiding this comment.
add an instance check; all the FSDataInputStream classes do that
There was a problem hiding this comment.
Ok, makes sense will do. Any objections if I just fix this here rather than filing a separate JIRA? The changes are pretty small / it would probably be less effort than creating, reviewing, and merging a separate patch just for this.
openinx
reviewed
Mar 21, 2019
| try { | ||
| decrypt(localDecryptor, localInBuffer, localOutBuffer, localPadding); | ||
| buf.position(start + len); | ||
| buf.limit(limit); |
There was a problem hiding this comment.
The limit should be the same as the start + len + Math.min(n - len, localInBuffer.remaining()) ?
There was a problem hiding this comment.
Well the invariant you want to preserve is that buf.put(inputBuf) should only be called with the original value of buf.limit() so that you don't exceed the given limit. Using start + len + Math.min(n - len, localInBuffer.remaining()) as the limit could violate this if n + start > buf.limit().
| * <p> | ||
| * In the case of an exception, the values of {@code buf.position()} and | ||
| * {@code buf.limit()} are undefined, and callers should be prepared to | ||
| * recover from this eventuality. |
There was a problem hiding this comment.
If an exception encountered , we should reset to the original pos & limit ? change the pos or limit is not friendly to the upper layer...
There was a problem hiding this comment.
I agree, but I copied this from ByteBufferReadable, so I think we should leave it for now, and if we want to lift this limitation, then we can do so for both ByteBufferReadable and ByteBufferPositionedReadable in another JIRA.
There was a problem hiding this comment.
Worth noting that the way it's implemented, it seems like on exception, the contents of the buffer are also undefined, right? ie we could have partially overwritten the buffer and then thrown?
There was a problem hiding this comment.
+1 for saying "buffer state undefined"
| * <p> | ||
| * Many implementations will throw {@link UnsupportedOperationException}, so | ||
| * callers that are not confident in support for this method from the | ||
| * underlying filesystem should be prepared to handle that exception. |
There was a problem hiding this comment.
we should provide an StreamCapabilities method for upstream to decide whether use the ByteBuffer pread , as we discussed in JIRA ?
There was a problem hiding this comment.
Yes, the code is there, updated the javadoc to reflect this.
| ByteBuffer buf) throws IOException { | ||
| int n = 0; | ||
| int total = 0; | ||
| while (n != -1) { |
There was a problem hiding this comment.
If n is 0, will we get stuck in a dead loop ? , try to use if(n > 0 )
There was a problem hiding this comment.
IIUC the HDFS read APIs make the same guarantees as InputStream#read(byte[], which returns -1 if there is no more data to read.
| @@ -316,6 +331,17 @@ void hdfsFileDisableDirectRead(hdfsFile file) | |||
| file->flags &= ~HDFS_FILE_SUPPORTS_DIRECT_READ; | |||
| } | |||
|
|
|||
| int hdfsFileUsesDirectPread(hdfsFile file) | |||
| { | |||
| return !!(file->flags & HDFS_FILE_SUPPORTS_DIRECT_PREAD); | |||
There was a problem hiding this comment.
It was copied from hdfsFileUsesDirectRead, but I agree it hard to understand so I cleaned it up.
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
toddlipcon
reviewed
Mar 27, 2019
| throws IOException { | ||
| checkStream(); | ||
| try { | ||
| int pos = buf.position(); |
There was a problem hiding this comment.
nit: I think it would be good to rename this pos to 'bufPos' so it's clearer that it's referring to the position in the buffer and not the current position in the file
| @@ -741,6 +830,7 @@ public boolean hasCapability(String capability) { | |||
| case StreamCapabilities.DROPBEHIND: | |||
| case StreamCapabilities.UNBUFFER: | |||
| case StreamCapabilities.READBYTEBUFFER: | |||
| case StreamCapabilities.PREADBYTEBUFFER: | |||
There was a problem hiding this comment.
it seems like preadByteBuffer requires the underlying stream to have this capability, so this should probably delegate to ((StreamCapabiltiies)in).hasCapability(PREADBYTEBUFFER), right?
(interestingly, the same goes for a few others of the capabilities like dropbehind, I think. Curious what @steveloughran has to say.
There was a problem hiding this comment.
if you forward to another stream, yes, wrap it. Theres's ongoing work in #575 to do more of this, with a matching PathCapabilities call for filesystems so you can see if a file you open under a path will have the feature before you open it (#568). These are both background bits of work so have languished neglected, but I'd love to see them in. StreamCapabilities is in hadoop 2.9+ BTW, so while the features you can look for evolves, at least the client code can compile without problems (and if you copy the new capabilities, link)
| } | ||
|
|
||
| return n; | ||
| } catch (ClassCastException e) { |
There was a problem hiding this comment.
hrm. I think we should probably do a follow-up JIRA to fix this, not for performance reasons, but because the try{...} block encompasses a lot of code. Let's say we accidentally screw up something in our encryption config and we get a ClassCastException somewhere inside decrypt. We'll swallow the real exception and claim that positioned read isn't supported, which isn't quite right.
So, I agree an instanceof check up front is probably the clearest from a code perspective and also avoids the above issue.
| return n; | ||
| } catch (ClassCastException e) { | ||
| throw new UnsupportedOperationException("This stream does not support " + | ||
| "positioned read."); |
There was a problem hiding this comment.
probably should specifically say "with byte buffers" or something
|
|
||
| /** | ||
| * Decrypt n bytes in buf starting at start. Output is also put into buf | ||
| * starting at current position. buf.position() and buf.limit() should be |
There was a problem hiding this comment.
This doc is confusing me a bit (haven't looked at the impl yet). It seems this both reads and writes to the same buf, but the read is happening from the 'start' whereas the write is happening at 'buf.position()'? That seems somewhat unexpected and opens up some questions about whether the output range and input range can overlap
There was a problem hiding this comment.
Separate nit: "should be unchanged" -> "will be unchanged" or "are not changed". Should be sounds awfully wishy-washy for a postcondition
There was a problem hiding this comment.
I think I copied this from the other #decrypt methods. Regardless, I rewrote this part so hopefully it is easier to understand.
| int len = 0; | ||
| Decryptor localDecryptor = null; | ||
| try { | ||
| localInBuffer = getBuffer(); |
There was a problem hiding this comment.
Can you add a TODO here that we can likely avoid one of these copies, at least when the byte buffer passed by the user is a direct buffer? It looks like the patch is currently doing:
pread -> user buffer
for each chunk:
copy from user buffer to tmp input
decrypt tmp input to tmp output
copy from tmp output to user buffer
but we could likely decrypt back to the user buffer directly, or decrypt from the user buffer to a tmp, and then write back. (this all assumes that the crypto codecs don't support in-place decryption, which they might)
10000
There was a problem hiding this comment.
looking briefly through openssl code, it seems like it actually supports in == out encryption, so maybe we could avoid both buffer copies for full blocks, and maybe avoid one buffer copy for the non-bytebuffer case as well by making inBuffer == outBuffer.
Again, probably just something to file for later
| ByteBuffer localInBuffer = null; | ||
| ByteBuffer localOutBuffer = null; | ||
| final int pos = buf.position(); | ||
| final int limit = buf.limit(); |
There was a problem hiding this comment.
instead of saving pos/limit here and restoring them later, would it be easier to duplicate() the bytebuffer? then you could easily just set the limit to match 'n' and not worry about it? The loop bounds might become a bit easier, too (while buf.remaining() > 0) etc since you no longer need to consider the passed-in length.
There was a problem hiding this comment.
I played around with this a bit, calling #duplicate avoids having to save and then reset the original limit and position, but I think you still have to set the limit when processing each chunk, other the call to localInBuffer.put(buf) will throw a BufferOverflowException. The Javadocs of #put say:
* If there are more bytes remaining in the
* source buffer than in this buffer, that is, if
* <tt>src.remaining()</tt> <tt>></tt> <tt>remaining()</tt>,
* then no bytes are transferred and a {@link
* BufferOverflowException} is thrown.
| * <p> | ||
| * In the case of an exception, the values of {@code buf.position()} and | ||
| * {@code buf.limit()} are undefined, and callers should be prepared to | ||
| * recover from this eventuality. |
There was a problem hiding this comment.
Worth noting that the way it's implemented, it seems like on exception, the contents of the buffer are also undefined, right? ie we could have partially overwritten the buffer and then thrown?
|
|
||
| /** | ||
| * Reads bytes using the read(ByteBuffer) API. By using Java | ||
| * DirectByteBuffers we can avoid copying the bytes from kernel space into |
There was a problem hiding this comment.
DirectByteBuffer avoids an extra copy into the java heap vs the C heap. It's still copying data out of the kernel to user space either way.
| * If the underlying stream supports the ByteBufferReadable interface then | ||
| * this method will transparently use read(ByteBuffer). This can help | ||
| * improve performance as it avoids unnecessary copies between the kernel | ||
| * space, the Java process space, and the C process space. |
There was a problem hiding this comment.
per above, the kernel->user transition's the same here, it's just avoiding some JVM heap copies
| Seekable, PositionedReadable, ByteBufferReadable, | ||
| ByteBufferPositionedReadable, HasFileDescriptor, CanSetDropBehind, | ||
| CanSetReadahead, HasEnhancedByteBufferAccess, ReadableByteChannel, | ||
| CanUnbuffer, StreamCapabilities { |
There was a problem hiding this comment.
adding one extra interface shouldn't force reformatting this entire set of lines -this only makes backporting on a common conflict point worse. Can you just add the new interface and the end and leave the rest alone? thanks
|
Addressed comments, rebased patch. |
hadoop-yetus
reviewed
Apr 2, 2019
| } | ||
| jthr = invokeMethod(env, &jVal, INSTANCE, jFile, | ||
| JC_FS_DATA_INPUT_STREAM, "hasCapability", "(Ljava/lang/String;)Z", | ||
| jCapabilityString); |
|
💔 -1 overall
This message was automatically generated. |
| Decryptor decryptor = null; | ||
| try { | ||
| // TODO we should be able to avoid copying chunks between the input buf, |
There was a problem hiding this comment.
I'd rather creating a new JIRA for this (now), over leaving a TODO in the code. You know that TODO is going to get forgotten about otherwise....
There was a problem hiding this comment.
Removed the TODO and filed HDFS-14417
|
Rebased a fixed a few checkstyle issues. |
hadoop-yetus
reviewed
Apr 6, 2019
| } | ||
| jthr = invokeMethod(env, &jVal, INSTANCE, jFile, | ||
| JC_FS_DATA_INPUT_STREAM, "hasCapability", "(Ljava/lang/String;)Z", | ||
| jCapabilityString); |
|
💔 -1 overall
This message was automatically generated. |
toddlipcon
reviewed
Apr 9, 2019
| @@ -327,7 +329,7 @@ public void close() throws IOException { | |||
| public int read(long position, byte[] buffer, int offset, int length) | |||
| throws IOException { | |||
| checkStream(); | |||
| try { | |||
| if (in instanceof PositionedReadable) { | |||
There was a problem hiding this comment.
nit: do you mind negating this check so that you can un-indent the rest of the method? ("early return" style)
| public int read(long position, final ByteBuffer buf) | ||
| throws IOException { | ||
| checkStream(); | ||
| if (in instanceof ByteBufferPositionedReadable) { |
There was a problem hiding this comment.
same here (and a couple similar below)
| * should be written to. | ||
| * <p> | ||
| * After a successful call, {@code buf.position()} will be advanced by the | ||
| * number of bytes read and {@code buf.limit()} should be unchanged. |
There was a problem hiding this comment.
"should be" -> "will be"
| @@ -40,8 +40,23 @@ | |||
|
|
|||
| // Bit fields for hdfsFile_internal flags | |||
| #define HDFS_FILE_SUPPORTS_DIRECT_READ (1<<0) | |||
| #define HDFS_FILE_SUPPORTS_DIRECT_PREAD (1<<0) | |||
There was a problem hiding this comment.
Yeah, thanks for catching that. I probably should not have just blindly copied the line above. I added a regression test for this.
| destroyLocalReference(env, jCapabilityString); | ||
| if (ret) { | ||
| errno = ret; | ||
| return 0; |
There was a problem hiding this comment.
seems a little weird to return 0 but also set errno here -- I see that you're doing it because you want to fall back to false on error, but then maybe errno should not be set
There was a problem hiding this comment.
Yeah. I think the reason for setting errno is that all the other methods set this to indicate that an exception occurred while the method was run. So if ret is true and errno needs to be set, then the method has to return with some value, right? Returning false seemed more reasonable than returning true.
| @@ -985,6 +1011,49 @@ int hdfsStreamBuilderSetDefaultBlockSize(struct hdfsStreamBuilder *bld, | |||
| return 0; | |||
| } | |||
|
|
|||
| static int hdfsHasStreamCapability(jobject jFile, | |||
There was a problem hiding this comment.
can you add a brief doc, even though this is just an internal method?
|
|
||
| //Error checking... make sure that this file is 'readable' | ||
| if (f->type != HDFS_STREAM_INPUT) { | ||
| fprintf(stderr, "Cannot read from a non-InputStream object!\n"); |
There was a problem hiding this comment.
seems odd to printf to stderr from a library. Is this our general pattern?
There was a problem hiding this comment.
Yeah, literally everything prints to stderr
| "preadDirect: FSDataInputStream#read"); | ||
| return -1; | ||
| } | ||
| return (jVal.i < 0) F438 ? 0 : jVal.i; |
There was a problem hiding this comment.
it seems the javadoc says that the method can return 0 for a transient zero-length result, and -1 for EOF. But here we're converting -1 to 0, so the caller can't distinguish between a must-retry '0 length read' and a 'shouldn't retry' EOF, right?
There was a problem hiding this comment.
Yeah, this looks like a latent bug. I double checked how hdfsRead works without the ByteBuffer optimization and it seems to be doing the right thing. I fixed this method and the pread equivalent to mimic what is done with the ByteBuffer optimization is disabled.
|
💔 -1 overall
This message was automatically generated. |
| } | ||
| jthr = invokeMethod(env, &jVal, INSTANCE, jFile, | ||
| JC_FS_DATA_INPUT_STREAM, "hasCapability", "(Ljava/lang/String;)Z", | ||
| jCapabilityString); |
|
💔 -1 overall
This message was automatically generated. |
9c5acf7 to
f5c692d
Compare
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
@toddlipcon addressed your comments. |
added 2 commits
April 29, 2019 15:42
* Remove the "try { ... } catch (ClassCatchException e)" pattern from
CryptoInputStream
* Fixed a bug in CryptoInputStream#hasCapability where the method was
not correctly delegating to the underlying stream
* Further refactoring of CryptoInputStream#decrypt methods to make the
code easier to understand
* Revised Javadoc for CryptoInputStream#decrypt
* Revised Javadoc for ByteBufferPositionedReadable and
ByteBufferReadable
* Revised Javadoc for readDirect, preadDirect, hdfsRead, and hdfsPread
|
💔 -1 overall
This message was automatically generated. |
|
Hadoop QA is still hitting these memory issues, so I ran the full suite of I looped I took a quick look at the heap dump as well and didn't see anything that was related to this patch. Interestingly, most of the memory is consumed by Mockito. |
|
Thanks. +1, merged to trunk. |
DremioQA
pushed a commit
to dremio/hadoop
that referenced
this pull request
Sep 4, 2019
Cherry-pick and fix conflicts for the following changes from trunk HDFS-14267. Add test_libhdfs_ops to libhdfs tests, mark libhdfs_read/write.c as examples. HDFS-14111. hdfsOpenFile on HDFS causes unnecessary IO from file offset 0. HDFS-3246: pRead equivalent for direct read path (apache#597) Change-Id: I883812e8bb5f0e951a2523336d63022ab61cf737
smengcl
pushed a commit
to smengcl/hadoop
that referenced
this pull request
Oct 8, 2019
Contributed by Sahil Takiar (cherry picked from commit 4877f0a) (cherry picked from commit 71cc468) Conflicts: hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ByteBufferReadable.java hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/TestCryptoStreams.java Change-Id: Ie80bb459e1e20d5ba1b39f6a281f8450765bf007
shanthoosh
pushed a commit
to shanthoosh/hadoop
that referenced
this pull request
Oct 15, 2019
Call for 'status' or 'kill' does not require Execution plan calculation. Author: Boris S <boryas@apache.org> Author: Boris Shkolnik <bshkolni@linkedin.com> Reviewers: Xinyu Liu <xinyuiscool@github.com> Closes apache#597 from sborya/RemoteAppRunnerStatus and squashes the following commits: 7e1feea0 [Boris S] retry 1c0b3e4f [Boris S] checkstyle e8d8d517 [Boris S] skipp graph planner for app status command 88f8559 [Boris S] Merge branch 'master' of https://github.com/apache/samza 0edf343 [Boris S] Merge branch 'master' of https://github.com/apache/samza 67e611e [Boris S] Merge branch 'master' of https://github.com/apache/samza dd39d08 [Boris S] Merge branch 'master' of https://github.com/apache/samza 1ad58d4 [Boris S] Merge branch 'master' of https://github.com/apache/samza 06b1ac3 [ 10000 Boris Shkolnik] Merge branch 'master' of https://github.com/sborya/samza 5e6f5fb [Boris Shkolnik] Merge branch 'master' of https://github.com/apache/samza 010fa16 [Boris S] Merge branch 'master' of https://github.com/apache/samza bbffb79 [Boris S] Merge branch 'master' of https://github.com/apache/samza d4620d6 [Boris S] Merge branch 'master' of https://github.com/apache/samza 410ce78 [Boris S] Merge branch 'master' of https://github.com/apache/samza a31a7aa [Boris Shkolnik] reduce debugging from info to debug in KafkaCheckpointManager.java
DremioQA
pushed a commit
to dremio/hadoop
that referenced
this pull request
Oct 24, 2019
Cherry-pick and fix conflicts for the following changes from trunk HDFS-14267. Add test_libhdfs_ops to libhdfs tests, mark libhdfs_read/write.c as examples. HDFS-14111. hdfsOpenFile on HDFS causes unnecessary IO from file offset 0. HDFS-3246: pRead equivalent for direct read path (apache#597) Change-Id: I883812e8bb5f0e951a2523336d63022ab61cf737 (cherry picked from commit 01c0d7d)
DremioQA
pushed a commit
to dremio/hadoop
that referenced
this pull request
Dec 27, 2019
Cherry-pick and fix conflicts for the following changes from trunk HDFS-14267. Add test_libhdfs_ops to libhdfs tests, mark libhdfs_read/write.c as examples. HDFS-14111. hdfsOpenFile on HDFS causes unnecessary IO from file offset 0. HDFS-3246: pRead equivalent for direct read path (apache#597) Change-Id: I883812e8bb5f0e951a2523336d63022ab61cf737 (cherry picked from commit 01c0d7d)
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Oct 25, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Oct 28, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
added a commit
to acceldata-io/hadoop
that referenced
this pull request
Oct 28, 2024
deepakdamri
added a commit
to acceldata-io/hadoop
that referenced
this pull request
Nov 20, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
added a commit
to acceldata-io/hadoop
that referenced
this pull request
Nov 27, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Nov 28, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Dec 2, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Dec 2, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Dec 2, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Dec 2, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
deepakdamri
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Dec 3, 2024
HDFS-3246: pRead equivalent for direct read path Contributed by Sahil Takiar
basic02
pushed a commit
to basic02/hadoop
that referenced
this pull request
Jan 12, 2025
Contributed by Sahil Takiar (cherry picked from commit 4877f0a) Conflicts: hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ByteBufferReadable.java hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/TestCryptoStreams.java Change-Id: Ie80bb459e1e20d5ba1b39f6a281f8450765bf007
harshith-212
added a commit
to acceldata-io/hadoop
that referenced
this pull request
Mar 5, 2025
This reverts commit 57202cc.
shubhluck
pushed a commit
to acceldata-io/hadoop
that referenced
this pull request
Mar 5, 2025
This reverts commit 57202cc.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
HDFS-3246: First several iterations of this patch are posted on the JIRA. This PR is a continuation of this work, it was created to make the code more reviewable. The version of the patch posted here fixes a few minor issues reported by Yetus, and added some more Javadocs to the changes in
CryptoInputStreamto make the code easier to understand.