pan3793 commented on code in PR #8511:
URL: https://github.com/apache/hadoop/pull/8511#discussion_r4476128722
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java:
##########
@@ -493,21 +498,108 @@ public void readVectored(final List<? extends FileRange>
ranges,
}
List<CombinedFileRange> checksumRanges = findChecksumRanges(dataRanges,
bytesPerSum, minSeek, maxSize);
- sums.readVectored(checksumRanges, allocate, release);
- datas.readVectored(dataRanges, allocate, release);
+ Map<FileRange, ByteBuffer> callerBuffers = new IdentityHashMap<>();
+ try {
+ // Allocate the caller's buffers synchronously, before submitting any
reads.
+ // Internal reads may be rounded to checksum boundaries; never return
slices
+ // of those buffers, as identity-tracking allocators cannot recycle
them.
+ for (FileRange range : sorted) {
+ ByteBuffer buffer = allocate.apply(range.getLength());
+ callerBuffers.put(range, buffer);
+ buffer.limit(range.getLength());
+ }
+ sums.readVectored(checksumRanges, allocate, release);
+ datas.readVectored(dataRanges, allocate, release);
+ } catch (IOException | RuntimeException | Error e) {
+ // No buffers have been exposed to the caller yet. Reads already
started
+ // may still be in flight, so release their buffers only after success.
+ releaseReadBuffers(checksumRanges, release);
+ releaseReadBuffers(dataRanges, release);
+ for (ByteBuffer buffer : callerBuffers.values()) {
+ releaseBuffer(buffer, release);
+ }
+ for (FileRange range : sorted) {
+ CompletableFuture<ByteBuffer> failed = new CompletableFuture<>();
+ failed.completeExceptionally(e);
+ range.setData(failed);
+ }
+ throw e;
+ }
for(CombinedFileRange checksumRange: checksumRanges) {
+ List<CompletableFuture<ByteBuffer>> verifications = new ArrayList<>();
for(FileRange dataRange: checksumRange.getUnderlying()) {
// when we have both the ranges, validate the checksum
CompletableFuture<ByteBuffer> result =
checksumRange.getData().thenCombineAsync(dataRange.getData(),
(sumBuffer, dataBuffer) ->
checkBytes(sumBuffer, checksumRange.getOffset(),
dataBuffer, dataRange.getOffset(), bytesPerSum,
file));
- // Now, slice the read data range to the user's ranges
- for(FileRange original: ((CombinedFileRange)
dataRange).getUnderlying()) {
- original.setData(result.thenApply(
- (b) -> VectoredReadUtils.sliceTo(b, dataRange.getOffset(),
original)));
+ verifications.add(result);
+ List<CompletableFuture<ByteBuffer>> copies = new ArrayList<>();
+ for (FileRange original : ((CombinedFileRange)
dataRange).getUnderlying()) {
+ ByteBuffer target = callerBuffers.get(original);
+ CompletableFuture<ByteBuffer> copy = result.thenApply(buffer -> {
+ target.put(VectoredReadUtils.sliceTo(buffer,
dataRange.getOffset(), original));
+ target.flip();
+ return target;
+ }).whenComplete((buffer, ex) -> {
+ if (ex != null) {
+ release.accept(target);
Review Comment:
Correction to my earlier comment: I said the data path was fine because
`whenComplete` adds the release failure as suppressed. It's attached to the
`CompletionException` wrapper, which `get()` discards, and
`FutureIO.awaitFuture` uses `get()`. Only `join()` keeps it. Please use
`releaseBuffer(target, release)` here so the failure is logged.
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java:
##########
@@ -493,21 +498,108 @@ public void readVectored(final List<? extends FileRange>
ranges,
}
List<CombinedFileRange> checksumRanges = findChecksumRanges(dataRanges,
bytesPerSum, minSeek, maxSize);
- sums.readVectored(checksumRanges, allocate, release);
- datas.readVectored(dataRanges, allocate, release);
+ Map<FileRange, ByteBuffer> callerBuffers = new IdentityHashMap<>();
+ try {
+ // Allocate the caller's buffers synchronously, before submitting any
reads.
+ // Internal reads may be rounded to checksum boundaries; never return
slices
+ // of those buffers, as identity-tracking allocators cannot recycle
them.
+ for (FileRange range : sorted) {
+ ByteBuffer buffer = allocate.apply(range.getLength());
+ callerBuffers.put(range, buffer);
+ buffer.limit(range.getLength());
+ }
+ sums.readVectored(checksumRanges, allocate, release);
Review Comment:
On the 2-arg path `release` is a no-op, so internal buffers taken from the
caller's allocator never go back, and an identity-tracking allocator still
reports them as leaked. They don't need to come from the caller. The 2-arg
overload could read `sums` and `datas` with an unpooled `ByteBuffer::allocate`
and a no-op release, while the 3-arg overload keeps `allocate` and `release`.
Every 2-arg `allocate` call is then a buffer the caller receives, which fixes
the leak from the JIRA for 2-arg callers without migrating to the 3-arg API.
The cost is small because these buffers are already effectively unpooled on
this path. In a standalone simulation of read, CRC and copy (single run, one
thread, M1 Max, page-cached file), heap callers were unchanged. Direct callers
went from 1250/3175/2360 to 1793/3318/2792 MB/s at 64 KB/1 MB/16 MB, because
unpooled `allocateDirect` is slower. The trade-off is that a direct caller's
internal buffers land on the heap, outside the allocator's limits.
The spec would need to say that `allocate()` isn't guaranteed to see every
allocation, and the two-argument paragraph at L744 would need updating. A test
should assert exactly one `allocate` call per range on the 2-arg path.
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java:
##########
@@ -493,21 +498,108 @@ public void readVectored(final List<? extends FileRange>
ranges,
}
List<CombinedFileRange> checksumRanges = findChecksumRanges(dataRanges,
bytesPerSum, minSeek, maxSize);
- sums.readVectored(checksumRanges, allocate, release);
- datas.readVectored(dataRanges, allocate, release);
+ Map<FileRange, ByteBuffer> callerBuffers = new IdentityHashMap<>();
+ try {
+ // Allocate the caller's buffers synchronously, before submitting any
reads.
+ // Internal reads may be rounded to checksum boundaries; never return
slices
+ // of those buffers, as identity-tracking allocators cannot recycle
them.
+ for (FileRange range : sorted) {
+ ByteBuffer buffer = allocate.apply(range.getLength());
+ callerBuffers.put(range, buffer);
+ buffer.limit(range.getLength());
+ }
+ sums.readVectored(checksumRanges, allocate, release);
+ datas.readVectored(dataRanges, allocate, release);
+ } catch (IOException | RuntimeException | Error e) {
+ // No buffers have been exposed to the caller yet. Reads already
started
+ // may still be in flight, so release their buffers only after success.
+ releaseReadBuffers(checksumRanges, release);
+ releaseReadBuffers(dataRanges, release);
+ for (ByteBuffer buffer : callerBuffers.values()) {
+ releaseBuffer(buffer, release);
+ }
+ for (FileRange range : sorted) {
+ CompletableFuture<ByteBuffer> failed = new CompletableFuture<>();
+ failed.completeExceptionally(e);
+ range.setData(failed);
+ }
+ throw e;
+ }
for(CombinedFileRange checksumRange: checksumRanges) {
+ List<CompletableFuture<ByteBuffer>> verifications = new ArrayList<>();
for(FileRange dataRange: checksumRange.getUnderlying()) {
// when we have both the ranges, validate the checksum
CompletableFuture<ByteBuffer> result =
checksumRange.getData().thenCombineAsync(dataRange.getData(),
(sumBuffer, dataBuffer) ->
checkBytes(sumBuffer, checksumRange.getOffset(),
dataBuffer, dataRange.getOffset(), bytesPerSum,
file));
- // Now, slice the read data range to the user's ranges
- for(FileRange original: ((CombinedFileRange)
dataRange).getUnderlying()) {
- original.setData(result.thenApply(
- (b) -> VectoredReadUtils.sliceTo(b, dataRange.getOffset(),
original)));
+ verifications.add(result);
+ List<CompletableFuture<ByteBuffer>> copies = new ArrayList<>();
+ for (FileRange original : ((CombinedFileRange)
dataRange).getUnderlying()) {
+ ByteBuffer target = callerBuffers.get(original);
+ CompletableFuture<ByteBuffer> copy = result.thenApply(buffer -> {
+ target.put(VectoredReadUtils.sliceTo(buffer,
dataRange.getOffset(), original));
+ target.flip();
+ return target;
+ }).whenComplete((buffer, ex) -> {
+ if (ex != null) {
+ release.accept(target);
+ }
+ });
+ original.setData(copy);
+ copies.add(copy);
}
+ // Wait for every copy before recycling the shared internal data
buffer.
+ CompletableFuture.allOf(copies.toArray(new CompletableFuture[0]))
Review Comment:
This cleanup depends on the futures handed to the caller, so cancelling a
range future runs it early. In a standalone reproduction of this chain with the
read still pending, `cancel()` blocked on `join()` until the read finished, the
internal data buffer was released before `checkBytes` ran, and the
`whenComplete` at L545 never ran, so `target` leaks. Nothing in Hadoop cancels
these futures, but internal cleanup shouldn't depend on them. Please run
cleanup on an internal stage and give the caller a separate future, releasing
`target` if that future was already cancelled.
##########
hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdatainputstream.md:
##########
@@ -669,13 +664,20 @@ Stream.hasCapability("in:readvectored")
[HADOOP-18296](https://issues.apache.org/jira/browse/HADOOP-18296),
_Memory fragmentation in ChecksumFileSystem Vectored IO implementation_
-highlights that `ChecksumFileSystem` (which the default implementation of
`file://`
-subclasses), may return buffers which are sliced subsets of buffers allocated
-through the `allocate()` function passed in.
-
-This will happen during reads with and without range coalescing.
-
-Checksum verification may be disabled by setting the option
+highlights that older versions of `ChecksumFileSystem` (which the default
implementation
+of `file://` subclasses) return sliced subsets of buffers allocated through the
+`allocate()` function passed in. These slices cannot be recycled by allocators
which
+track buffer identity and can retain oversized backing allocations.
+
+With [HADOOP-19901](https://issues.apache.org/jira/browse/HADOOP-19901),
+`ChecksumFileSystem` verifies data in internal buffers and copies
+each requested range into its own buffer, allocated synchronously through
`allocate()`.
+It returns exactly that buffer, not a slice, and reports `false` for
+`fs.capability.vectoredio.sliced` on both the filesystem and the stream.
+This applies with and without checksum verification, so direct buffers and
+identity-tracking allocators can be used without a slicing workaround.
+
+For older releases, checksum verification may be disabled by setting the option
Review Comment:
`core-default.xml` also needs updating. The `fs.file.checksum.verify`
description ends with "When fs.file.checksum.verify is false, vector reads of
data will always return buffers that are the buffers allocated through the
buffer allocator passed in to the API call and not sliced subsets thereof."
Buffers are now never sliced, so that sentence should go.
--
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]
