anmolanmol1234 commented on code in PR #8830:
URL: https://github.com/apache/hadoop/pull/8830#discussion_r4522564423
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferManagerV2.java:
##########
@@ -279,16 +276,135 @@ void init() {
}
/**
- * {@link AbfsInputStream} calls this method to queueing read-ahead.
- * @param stream which read-ahead is requested from.
- * @param requestedOffset The offset in the file which should be read.
- * @param requestedLength The length to read.
+ * {@inheritDoc}
*/
@Override
public void queueReadAhead(final AbfsInputStream stream,
final long requestedOffset,
final int requestedLength,
+ final List<BlobLayout.BlobRange> segments,
TracingContext tracingContext) {
+ synchronized (this) {
+ if (isAlreadyQueued(stream.getETag(), requestedOffset)) {
+ // Already queued for this offset, so skip queuing.
+ printTraceLog(
+ "Skipping queuing readAhead for file: {}, with eTag: {}, "
+ + "offset: {}, triggered by stream: {} as it is already
queued",
+ stream.getPath(), stream.getETag(), requestedOffset,
+ stream.hashCode());
+ return;
+ }
+
+ if (isFreeListEmpty() && !tryMemoryUpscale() && !tryEvict()) {
+ // No buffers are available and more buffers cannot be created. Skip
queuing.
+ printTraceLog(
+ "Skipping queuing readAhead for file: {}, with eTag: {}, offset:
{}, triggered by stream: {} as no buffers are available",
+ stream.getPath(), stream.getETag(), requestedOffset,
+ stream.hashCode());
+ return;
+ }
+
+ // Create new parent for this window
+ ReadBuffer parentBuffer = new ReadBuffer();
+ // To map buffer with stream that requested it
+ parentBuffer.setStream(stream);
+ // To map buffer with file it belongs to
+ parentBuffer.setETag(stream.getETag());
+ parentBuffer.setPath(stream.getPath());
+ parentBuffer.setOffset(requestedOffset);
+ parentBuffer.setBufferOffset(0);
+ parentBuffer.setLength(0);
+ parentBuffer.setStatus(ReadBufferStatus.NOT_AVAILABLE);
+ parentBuffer.setTracingContext(tracingContext);
+ parentBuffer.setRequestedLength(requestedLength);
+ parentBuffer.setParentToTrue();
+
+ if (isFreeListEmpty()) {
+ /*
+ * By now there should be at least one buffer available.
+ * This is to double sure that after upscaling or eviction,
+ * we still have free buffer available. If not, we skip queueing.
+ */
+ return;
+ }
+ Integer bufferIndex = popFromFreeList();
+ if (bufferIndex >= bufferPool.length) {
+ // This should never happen.
+ printTraceLog(
+ "Skipping queuing readAhead for file: {}, with eTag: {}, offset:
{}, triggered by stream: {} as invalid buffer index popped from free list",
+ stream.getPath(), stream.getETag(), requestedOffset,
+ stream.hashCode());
+ return;
+ }
+ parentBuffer.setBuffer(bufferPool[bufferIndex]);
+ parentBuffer.setBufferindex(bufferIndex);
+
+ printTraceLog(
+ "Created parent buffer for offset: {}, with index: {}, for file: {},
eTag: {}",
+ requestedOffset, bufferIndex, stream.getPath(), stream.getETag());
+
+ // Iterate over the segments and create children
+ for (BlobLayout.BlobRange segment : segments) {
+ long readStart = segment.start();
+ long readEnd = segment.end();
+ int rangeLength = Math.toIntExact(readEnd - readStart + 1);
+
+ printTraceLog(
+ "Start Queueing readAhead childBuffer for file: {}, with eTag: {},
"
+ + "offset: {}, length: {}, segments: {}, triggered by stream:
{}",
+ stream.getPath(), stream.getETag(), readStart, rangeLength,
+ segments.size(), stream.hashCode());
+
+ // Create child buffer for this segment
+ ReadBuffer childBuffer = new ReadBuffer();
+ childBuffer.setStream(stream);
+ childBuffer.setETag(
+ stream.getETag()); // To map buffer with file it belongs to
+ childBuffer.setPath(stream.getPath());
+ childBuffer.setOffset(readStart); // File position for this segment
+ childBuffer.setBufferOffset(
+ parentBuffer.getBufferOffset()); // Offset in shared array
+ childBuffer.setLength(0);
+ childBuffer.setRequestedLength(rangeLength);
+ childBuffer.setStatus(ReadBufferStatus.NOT_AVAILABLE);
+ childBuffer.setLatch(new CountDownLatch(1));
+ childBuffer.setTracingContext(tracingContext);
+
+ childBuffer.setBuffer(parentBuffer.getBuffer()); // Share parent's
array
+ childBuffer.setBufferindex(
+ parentBuffer.getBufferindex()); // Share parent's index
+ childBuffer.setParentBuffer(parentBuffer); // Link to parent
+ childBuffer.setReadEndpoint(segment.host());
+
+ parentBuffer.pushToChildBufferList(childBuffer);
+ parentBuffer.incrementActiveChildren();
+ parentBuffer.setBufferOffset(
Review Comment:
Child buffers are placed one after another in the shared array, but
getBlockFromCompletedQueue reads from position \- parentOffset. That only works
if the ranges are contiguous and the first one starts at requestedOffset.
getBlobRanges can return partial coverage, for example when a layout fetch
times out or fails, and readInternal only falls back when the list is empty.
Example: for a window of 0\-4 MB where only [1 MB\-4 MB) is returned, the child
writes file bytes from 1 MB into array index 0. A read at position 0 then
silently returns the wrong data. With a gap in the middle, a read can hit
unfilled or stale pool bytes.
If the service can return overlapping ranges from different hosts
(BlobLayout merges only same\-host ranges), the summed lengths could also
overflow the shared buffer.
Can we set each child's buffer offset to readStart \- requestedOffset? If
the ranges don't fully and contiguously cover the window, we should fall back
to the single\-buffer path.
--
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]
