[
https://issues.apache.org/jira/browse/HADOOP-19901?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18137103#comment-18137103
]
ASF GitHub Bot commented on HADOOP-19901:
-----------------------------------------
pan3793 commented on code in PR #8511:
URL: https://github.com/apache/hadoop/pull/8511#discussion_r4452981333
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java:
##########
@@ -493,22 +493,67 @@ 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);
+ try {
+ sums.readVectored(checksumRanges, allocate, release);
+ datas.readVectored(dataRanges, allocate, release);
+ } catch (IOException | RuntimeException 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);
+ 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));
+ dataBuffer, dataRange.getOffset(), bytesPerSum,
file))
+ .whenComplete((buffer, ex) -> {
+ // Failed reads release their own buffers. A successful
data
+ // read whose verification failed is never returned to the
caller.
+ if (ex != null &&
!dataRange.getData().isCompletedExceptionally()) {
+ release.accept(dataRange.getData().join());
+ }
+ });
+ verifications.add(result);
// 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)));
Review Comment:
`sliceTo` always returns a new slice, so under `ChecksumFileSystem` the
caller never gets back the buffer its allocator returned (`RawLocalFileSystem`
returns the buffer itself). An allocator that tracks buffers by identity (e.g.
Parquet's `TrackingByteBufferAllocator`) still can't release data buffers here.
The spec already treats this as a defect: the "Buffer Slicing" section in
`fsdatainputstream.md` (HADOOP-18296) says "Don't slice buffers.
`ChecksumFileSystem` has to be considered an outlier which needs to be
addressed in future". The new javadoc on `PositionedReadable` ("or if it or a
slice of it has been returned") turns that outlier into a general rule.
Also, the 2-arg overload keeps the no-op release, so a Parquet caller on the
2-arg API still leaks checksum buffers with this patch. If Parquet moves to the
3-arg API but keeps its capturing-allocator workaround, checksum buffers are
released twice.
Could this PR fix data-buffer ownership as well, so the "or a slice of it"
clause can be dropped? This follows what @steveloughran proposed on
HADOOP-18296:
1. Read data ranges into internal buffers, as checksum buffers are read now.
After verification, copy each caller range into a buffer from
`allocate.apply(original.getLength())`, allocated synchronously in
`readVectored`, and release the internal buffer. Every buffer the caller sees
is then exactly one allocation, and the cost is one extra copy on a path that
already computes CRC32 over every byte. I/O count, coalescing and async
behavior are unchanged. This is consistent with the vectored-read spec, which
covers range coalescing, async and parallel reads, and caller-controlled
allocation, and does not promise zero-copy. Returning exactly the allocated
buffers serves the caller-controlled-allocation part better than slices do.
2. If `VectoredReadBenchmark` shows the copy matters, a zero-copy
alternative is to read only the head and tail partial chunks into small
internal buffers (at most 512 bytes each) and the caller's range straight into
its own buffer, then verify across the pieces with sequential `CRC32.update`
calls. Merging doesn't matter here since `maxReadSizeForVectorReads()` is 0.
With either, `hasCapability(VECTOREDIO_BUFFERS_SLICED)` (L541, L1196) should
return false, and the "Buffer Slicing" section in the spec needs updating.
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/PositionedReadable.java:
##########
@@ -141,16 +141,20 @@ default void readVectored(List<? extends FileRange>
ranges,
/**
* Extension of {@link #readVectored(List, IntFunction)} where a {@code
release(buffer)}
- * operation may be invoked if problems surface during reads.
+ * operation may be invoked if problems surface during reads, or to release
+ * internal buffers which are not returned to the caller.
* <p>
- * The {@code release} operation is invoked after an IOException
- * to return the actively buffer to a pool before reporting a failure
- * in the future.
+ * For buffers holding caller-visible range data, the {@code release}
operation
+ * is invoked after an IOException to return the active buffer to a pool
before
+ * reporting a failure in the future. Buffers used only internally, such as
+ * checksum buffers, should be released once they are no longer needed,
whether
+ * the read succeeds or fails. For each allocation, a buffer must not be
released
+ * more than once, or if it or a slice of it has been returned to the caller.
Review Comment:
With the cleanup at `ChecksumFileSystem.java:499`, `release` can be invoked
after `readVectored` has thrown, once in-flight internal reads complete. Please
state that here and in `fsdatainputstream.md`, e.g. "`release` may be invoked
asynchronously, including after this method has thrown; callers must keep the
release target valid until all submitted reads complete."
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java:
##########
@@ -493,22 +493,67 @@ 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);
+ try {
+ sums.readVectored(checksumRanges, allocate, release);
+ datas.readVectored(dataRanges, allocate, release);
+ } catch (IOException | RuntimeException 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);
+ 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));
+ dataBuffer, dataRange.getOffset(), bytesPerSum,
file))
+ .whenComplete((buffer, ex) -> {
+ // Failed reads release their own buffers. A successful
data
+ // read whose verification failed is never returned to the
caller.
+ if (ex != null &&
!dataRange.getData().isCompletedExceptionally()) {
+ release.accept(dataRange.getData().join());
+ }
+ });
+ verifications.add(result);
// 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)));
}
}
+ // Release the checksum buffer once all verifications using it are
complete.
+ // Without this, buffers allocated through the caller's allocator for
checksum
+ // data are never released, since the caller has no reference to them.
+ CompletableFuture.allOf(verifications.toArray(new
CompletableFuture[0]))
+ .whenComplete((unused, ex) -> {
+ // A failed checksum read already releases its buffer in the
inner stream.
+ // Otherwise release it even if verification or a data read
failed.
+ if (!checksumRange.getData().isCompletedExceptionally()) {
+ release.accept(checksumRange.getData().join());
Review Comment:
This `whenComplete` stage and the `thenAccept` at L555 are discarded, so an
exception from the caller's `release` is lost without a trace. The data-buffer
path is fine because `whenComplete` adds it as suppressed to the failure the
caller sees. Please catch it and `LOG.warn` here and at L555.
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java:
##########
@@ -414,10 +424,22 @@ private static class AsyncHandler implements
CompletionHandler<Integer, Integer>
private void initiateRead() {
for(int i = 0; i < ranges.size(); ++i) {
FileRange range = ranges.get(i);
- buffers[i] = allocateRelease.getBuffer(false, range.getLength());
- final ByteBuffer buffer = buffers[i];
- LOG.debug("Reading file range {} into buffer {}", range,
System.identityHashCode(buffer));
- channel.read(buffer, range.getOffset(), i, this);
+ try {
+ buffers[i] = allocateRelease.getBuffer(false, range.getLength());
+ final ByteBuffer buffer = buffers[i];
+ // Pooled buffers may be larger than the requested range.
+ buffer.limit(range.getLength());
+ LOG.debug("Reading file range {} into buffer {}", range,
System.identityHashCode(buffer));
+ channel.read(buffer, range.getOffset(), i, this);
+ } catch (RuntimeException e) {
Review Comment:
`allocateDirect` reports exhaustion as `OutOfMemoryError("Direct buffer
memory")`, which skips this handler. The remaining futures are never completed.
Catching `RuntimeException | Error` (rethrowing as is) would cover it.
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java:
##########
@@ -493,22 +493,67 @@ 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);
+ try {
+ sums.readVectored(checksumRanges, allocate, release);
+ datas.readVectored(dataRanges, allocate, release);
+ } catch (IOException | RuntimeException e) {
Review Comment:
Same here: an `Error` from the checksum or data submission bypasses
`releaseReadBuffers`, so buffers from reads already submitted are never
released.
##########
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java:
##########
@@ -349,8 +349,18 @@ public void readVectored(final List<? extends FileRange>
ranges,
range.setData(new CompletableFuture<>());
}
final ByteBufferPool pool = new VectorIOBufferPool(allocate, release);
+ final AsynchronousFileChannel channel;
+ try {
+ channel = getAsyncChannel();
Review Comment:
Pre-existing, not introduced here: `getAsyncChannel()` (L321) uses broken
double-checked locking. `asyncChannel` is not volatile and there is no null
re-check inside `synchronized`. Two concurrent `readVectored` calls on the same
stream can each open an `AsynchronousFileChannel`, and the overwritten one is
never closed. Fine as a separate JIRA.
> 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]
