[
https://issues.apache.org/jira/browse/HADOOP-19901?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18138469#comment-18138469
]
ASF GitHub Bot commented on HADOOP-19901:
-----------------------------------------
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.
> ChecksumFileSystem.readVectored leaks buffers allocated through caller's
> IntFunction allocator
> ----------------------------------------------------------------------------------------------
>
> Key: HADOOP-19901
> URL: https://issues.apache.org/jira/browse/HADOOP-19901
> Project: Hadoop Common
> Issue Type: Bug
> Components: fs
> Affects Versions: 3.5.0, 3.4.3
> Reporter: Ismaël Mejía
> Priority: Major
> Labels: pull-request-available
>
> h3. Summary
> When {{ChecksumFileSystem.readVectored()}} is called with checksum
> verification enabled (the default for {{LocalFileSystem}}), it allocates
> buffers for *both* file data ranges and checksum ranges through the
> caller-provided {{IntFunction<ByteBuffer> allocate}} function. However, the
> checksum buffers are only used temporarily for verification and are never
> released back to the caller. The caller has no reference to these buffers and
> no mechanism to release them.
> This was discovered in Apache Parquet Java while upgrading from Hadoop 3.3.0
> to 3.4.3 and testing with {{TrackingByteBufferAllocator}}, which detected
> leaked {{ByteBuffer}} allocations.
> h3. Root cause
> In {{ChecksumFSInputChecker.readVectored()}} ([ChecksumFileSystem.java,
> trunk|https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java]):
> {code:java}
> @Override
> public void readVectored(final List<? extends FileRange> ranges,
> final IntFunction<ByteBuffer> allocate,
> final Consumer<ByteBuffer> release) throws IOException {
> // ...
> sums.readVectored(checksumRanges, allocate, release); // allocates
> checksum buffers via caller's allocator
> datas.readVectored(dataRanges, allocate, release); // allocates data
> buffers via caller's allocator
> for (CombinedFileRange checksumRange : checksumRanges) {
> for (FileRange dataRange : checksumRange.getUnderlying()) {
> CompletableFuture<ByteBuffer> result =
> checksumRange.getData().thenCombineAsync(dataRange.getData(),
> (sumBuffer, dataBuffer) ->
> checkBytes(sumBuffer, checksumRange.getOffset(),
> dataBuffer, dataRange.getOffset(), bytesPerSum, file));
> for (FileRange original : ((CombinedFileRange)
> dataRange).getUnderlying()) {
> original.setData(result.thenApply(
> (b) -> VectoredReadUtils.sliceTo(b, dataRange.getOffset(),
> original)));
> }
> }
> }
> }
> {code}
> Two problems:
> # *Checksum buffers are never released.* {{sums.readVectored(checksumRanges,
> allocate, release)}} allocates buffers through the caller's {{allocate}}
> function to read checksum data. After {{checkBytes()}} verifies the data, the
> checksum buffers ({{sumBuffer}}) are no longer needed, but they are never
> passed to {{release}} and are invisible to the caller. They leak.
> # *The 2-arg API provides no release mechanism.* The 2-arg overload passes a
> no-op release:
> {code:java}
> public void readVectored(List<? extends FileRange> ranges,
> IntFunction<ByteBuffer> allocate) throws
> IOException {
> readVectored(ranges, allocate, (b) -> { });
> }
> {code}
> Even callers using the 3-arg API don't benefit, because
> {{ChecksumFileSystem}} itself never calls {{release}} on the checksum buffers
> -- it only passes {{release}} down to the underlying streams.
> h3. How this was discovered
> Apache Parquet Java uses a {{TrackingByteBufferAllocator}} in tests that
> wraps the real allocator and tracks all allocations. When the allocator is
> closed, it throws {{LeakedByteBufferException}} if any allocated buffers were
> not released. After upgrading Hadoop from 3.3.0 to 3.4.3, the following test
> classes started failing with buffer leak errors in the vectored I/O path:
> * {{TestRecordLevelFilters}} (15 tests)
> * {{TestColumnIndexFiltering}} (24 tests)
> * {{TestParquetReader}} (6+ tests)
> The allocation stacktrace showed:
> {code}
> TrackingByteBufferAllocator.allocate
> -> VectorIOBufferPool.getBuffer
> -> RawLocalFileSystem$AsyncHandler.initiateRead
> {code}
> Parquet's {{readVectored()}} method passes a {{ByteBufferAllocator}} to
> Hadoop, but Hadoop uses it for internal temporary allocations (checksum
> ranges) that are invisible to the caller.
> h3. Workaround in Parquet
> We implemented a "capturing allocator" pattern that wraps the allocator to
> track all buffers allocated during {{readVectored()}}, then registers them
> all for release:
> {code:java}
> List<ByteBuffer> allocatedBuffers = new ArrayList<>();
> ByteBufferAllocator capturingAllocator = new ByteBufferAllocator() {
> @Override
> public ByteBuffer allocate(int size) {
> ByteBuffer buf = options.getAllocator().allocate(size);
> allocatedBuffers.add(buf);
> return buf;
> }
> // ...
> };
> try {
> f.readVectored(ranges, capturingAllocator);
> // ... process futures ...
> } finally {
> builder.addBuffersToRelease(allocatedBuffers);
> }
> {code}
> This ensures all buffers allocated through the caller's allocator are
> eventually released, regardless of whether they are returned in a future or
> used internally by ChecksumFileSystem. See [parquet-java commit
> fc0586d68|https://github.com/apache/parquet-java/commit/fc0586d68].
> h3. Suggested fixes
> *Option A (minimal): Release checksum buffers after verification.*
> In {{ChecksumFSInputChecker.readVectored()}}, after {{checkBytes()}}
> completes, call {{release}} on the checksum buffer:
> {code:java}
> CompletableFuture<ByteBuffer> result =
> checksumRange.getData().thenCombineAsync(dataRange.getData(),
> (sumBuffer, dataBuffer) -> {
> ByteBuffer verified = checkBytes(sumBuffer,
> checksumRange.getOffset(),
> dataBuffer, dataRange.getOffset(), bytesPerSum, file);
> release.accept(sumBuffer); // release checksum buffer after
> verification
> return verified;
> });
> {code}
> *Option B (comprehensive): Don't use the caller's allocator for internal
> temporaries.*
> ChecksumFileSystem should allocate its own temporary buffers for checksum
> data instead of using the caller-provided allocator. The caller's allocator
> is intended for buffers that the caller will own and manage. Using it for
> internal temporaries violates that expectation.
> {code:java}
> // Use internal allocation for checksums, not the caller's allocator
> sums.readVectored(checksumRanges, ByteBuffer::allocate, (b) -> { });
> // Only use caller's allocator for data ranges
> datas.readVectored(dataRanges, allocate, release);
> {code}
> *Option C (API improvement): Extend the API to support paired
> allocate/release.*
> The current {{IntFunction<ByteBuffer>}} allocator is one-way -- there's no
> way for Hadoop to release a buffer it allocated through the caller's
> function. HADOOP-19303 added a {{Consumer<ByteBuffer> release}} parameter,
> but it's separate from the allocate function and {{ChecksumFileSystem}}
> doesn't use it for its own intermediate buffers. A paired allocator/releaser
> interface (similar to Parquet's {{ByteBufferAllocator}} with both
> {{allocate}} and {{release}} methods) would make the lifecycle explicit.
> h3. Related issues
> * *HADOOP-19303* (VectorIO API to support releasing buffers on failure) --
> Added the 3-arg {{readVectored}} with {{release}} Consumer, but
> {{ChecksumFileSystem}} doesn't call {{release}} on checksum buffers.
> * *HADOOP-18296* (Memory fragmentation in ChecksumFileSystem Vectored IO) --
> Fixed range merging fragmentation, but did not address checksum buffer leaks.
> * *PARQUET-2171* (Implement vectored IO in parquet file format) -- The
> Parquet side implementation.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
