HDDS-15734. Implementing position read in BlockInputStream#10765
HDDS-15734. Implementing position read in BlockInputStream#10765chungen0126 wants to merge 7 commits into
Conversation
|
cc @yandrey321 |
Bugbot reviewreadFully succeeds at EOF —
Positioned read leaks clients —
readFully checks buffer capacity —
Refresh skipped after first failure — After the first refresh attempt for a pipeline, Generated-by: Cursor Bugbot |
|
|
||
| private BlockData blockData; | ||
|
|
||
| private Pipeline failedPipeline; |
There was a problem hiding this comment.
why do we need to track previously failed pipelines?
There was a problem hiding this comment.
The reason I track previously failed pipelines is that, in my design, I assume a scenario where multiple position read in BlockInputStream might fail concurrently and attempt to refresh the block info at the same time. Once the first BlockInputStream completes the refresh, subsequent streams will obtain the updated information. If the position read detects that its failed pipeline belongs to the old one and a new pipeline is now available, it should not need to trigger another redundant refresh.
| int len = buffer.remaining(); | ||
| int innerRetries = 0; | ||
| int chunkIdx = Arrays.binarySearch(chunkOffsets, pos); | ||
| if (chunkIdx < 0) { |
There was a problem hiding this comment.
when index can be negative?
There was a problem hiding this comment.
If the binary search doesn't find an exact match in the array, it returns a negative number. For details, please see: https://docs.oracle.com/javase/8/docs/api/java/util/Arrays.html#binarySearch-int:A-int-
| final List<ChunkInputStream> inputStreams = this.chunkStreams; | ||
| if (inputStreams != null) { | ||
| for (ChunkInputStream is : inputStreams) { | ||
| is.releaseClient(); |
There was a problem hiding this comment.
should it be wrapped in try catch?
There was a problem hiding this comment.
I don't think it's necessary. They didn't throw any exceptions.
| } | ||
|
|
||
| @Test | ||
| public void testPositionedRead() throws Exception { |
There was a problem hiding this comment.
these should cover moving forward and back in the file.
| ByteBuffer byteBuffer = ByteBuffer.wrap(buffer); | ||
| int bytesRead = chunkStream.read(30, byteBuffer); | ||
|
|
||
| assertEquals(50, bytesRead); |
There was a problem hiding this comment.
what if chunk stream has less than 50 bytes?
| } | ||
|
|
||
| @Test | ||
| public void testPositionedReadFully() throws Exception { |
There was a problem hiding this comment.
how do we assert that stream was read till the end as text name suggest?
3 participants
What changes were proposed in this pull request?
Summary
To resolve the significant lock contention experienced during concurrent readVectored operations, this PR implements the ByteBufferPositionedReadable interface for both BlockInputStream and ChunkInputStream. By supporting thread-safe, position-based reads directly into ByteBuffer, we bypass the synchronization overhead of traditional stream locks and greatly improve parallel read performance.
Detailed Changes
OzoneFSInputStreamto leverage the positional read capabilities ofBlockInputStream. This allowsOzoneFSInputStreamto delegate position-based reads directly to the underlying block stream, bypassing global stream-level locks and enabling fully concurrent read operations.What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-15734
How was this patch tested?
Add test in
TestChunkInputStreamandTestBlockInputStream. Also test byTestOzoneFSInputStream.CI: https://github.com/chungen0126/ozone/actions/runs/29289303279