iemejia commented on code in PR #8511:
URL: https://github.com/apache/hadoop/pull/8511#discussion_r4470497304
##########
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:
Thanks for the follow-up review. Addressed in
a419df0d808e80fe6d146089dada638fa0c8f057 using the copy-after-verification
approach.
`ChecksumFileSystem` now allocates each caller-range buffer synchronously in
`readVectored()`, copies the verified bytes from the internal data buffer, and
returns the exact allocated object rather than a slice. Internal data buffers
are released after all dependent copies finish, including on failure. Both the
stream and filesystem now report `false` for `VECTOREDIO_BUFFERS_SLICED`, and
the “Buffer Slicing” section and the proposed release-contract wording have
been updated to remove the slice clause.
The two-argument overload retains its no-op release behavior. The docs now
explain that explicit recycling of internal buffers requires the three-argument
overload, and that callers migrating from a capturing-allocator workaround must
not separately release buffers already released through the callback. Only
successfully returned range buffers remain caller-owned.
Regression coverage checks exact buffer identity, synchronous allocation,
unaligned ranges, oversized pooled buffers, copy failures, and preservation of
successful sibling results with both heap and direct buffers.
All **215 targeted test cases pass with JDK 17** across
`TestLocalFSContractVectoredRead`, `TestRawLocalContractVectoredRead`,
`TestVectoredReadUtils`, `TestChecksumFileSystem`, and
`TestRawLocalFileSystemVectoredRead`. Checkstyle reports no violations on
changed lines, and `git diff --check` passes.
##########
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:
Addressed in a419df0d808e80fe6d146089dada638fa0c8f057. The checksum/data
setup handler now also catches `Error` and arranges cleanup for internal reads
already submitted. Their buffers are released only after successful read
completion; unsubmitted caller buffers are released immediately, and the
original setup failure is preserved.
Tests cover injected errors during caller, checksum, and data allocation,
plus partial checksum/data submission with both pending and immediately
completed reads. Caller-range futures are also completed exceptionally on setup
failure.
##########
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:
Addressed in a419df0d808e80fe6d146089dada638fa0c8f057. Cleanup that has no
caller-visible future now goes through a helper which catches release-callback
failures and calls `LOG.warn`, rather than losing them in discarded completion
stages or masking the original synchronous setup failure.
This covers checksum cleanup, the new internal data-buffer cleanup after
copying, and cleanup of successful internal reads after synchronous setup
fails. Regression tests verify logging for checksum/data cleanup and
setup-failure cleanup, while retaining exact release-count and buffer-identity
assertions.
##########
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:
Addressed in a419df0d808e80fe6d146089dada638fa0c8f057. Both
`PositionedReadable` and `fsdatainputstream.md` now explicitly state that
`release` may be invoked asynchronously, including after `readVectored()` has
thrown, and that callers must keep the release target valid until all submitted
reads complete.
##########
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:
Addressed in a419df0d808e80fe6d146089dada638fa0c8f057. Allocation and
initial submission now catch `RuntimeException | Error`, release the affected
allocated buffer, settle unsubmitted range futures, and rethrow the original
failure. Earlier submitted reads retain ownership until their callbacks finish.
Channel-open failures and partial-read resubmissions also handle `Error`.
Deterministic tests inject `OutOfMemoryError("Direct buffer memory")` without
exhausting memory and cover both pending and immediately completed earlier
reads, as well as throwing release callbacks.
##########
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:
Agreed. I have left the pre-existing async-channel initialization issue
unchanged in this PR so it can be handled in a separate JIRA, as suggested.
--
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]
