[
https://issues.apache.org/jira/browse/HADOOP-19946?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141841#comment-18141841
]
ASF GitHub Bot commented on HADOOP-19946:
-----------------------------------------
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.
> ABFS: Enable data locality aware reads for Azure Blob Storage
> -------------------------------------------------------------
>
> Key: HADOOP-19946
> URL: https://issues.apache.org/jira/browse/HADOOP-19946
> Project: Hadoop Common
> Issue Type: New Feature
> Components: fs/azure
> Reporter: Manish Bhatt
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
>
> Introduce data locality support in ABFS to improve read performance for Azure
> Blob Storage by routing read requests to the endpoint that hosts the required
> data range. The goal is to reduce latency, improve throughput, and make read
> operations more efficient for large files and distributed workloads. This
> work adds blob layout discovery, range-to-endpoint mapping, locality-aware
> range selection, and caching of layout metadata to avoid repeated lookups. It
> also enables segmented read-ahead aligned with locality boundaries and adds
> safeguards for concurrency, eviction, timeout handling, and failure
> propagation needed for reliable production-scale reads.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
