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.
--
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]
