Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Avoid get block info when positioned read for first retry #18417

Merged
merged 12 commits into from
Jan 12, 2024
Merged

Avoid get block info when positioned read for first retry #18417

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
625955e
Avoid get block info when positioned read for first retry
maobaolong Nov 15, 2023
3752348
Address comments, add expiration time
maobaolong Dec 4, 2023
5d24946
Fix style
maobaolong Dec 8, 2023
ae3a997
Update core/common/src/main/java/alluxio/conf/PropertyKey.java
maobaolong Dec 11, 2023
e8dc5dc
Add UT
maobaolong Dec 22, 2023
33ed1f7
Add UT
maobaolong Dec 22, 2023
d428455
style
maobaolong Dec 24, 2023
b24a7f4
Fix test
maobaolong Dec 25, 2023
11f1d84
Fix test
maobaolong Dec 27, 2023
a2a04a3
Fix test2
maobaolong Jan 2, 2024
d59a6a1
Add description
maobaolong Jan 12, 2024
e6960f0
Add description
maobaolong Jan 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions core/client/fs/src/main/java/alluxio/client/file/AlluxioFileInStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,11 +302,15 @@ private int positionedReadInternal(long pos, byte[] b, int off, int len) throws
try {
// Positioned read may be called multiple times for the same block. Caching the in-stream
// allows us to avoid the block store rpc to open a new stream for each call.
BlockInfo blockInfo = lastException == null
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add a threshold of outdate time

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updateStream check outdate time

? mStatus.getBlockInfo(blockId) : mBlockStore.getInfo(blockId);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dbw9580 do you think it's possible that we can use mStatus.getBlockInfo(blockId) even if lastException != null?

if (mCachedPositionedReadStream == null) {
mCachedPositionedReadStream = mBlockStore.getInStream(blockId, mOptions, mFailedWorkers);
mCachedPositionedReadStream = mBlockStore.getInStream(
blockInfo, mOptions, mFailedWorkers);
} else if (mCachedPositionedReadStream.getId() != blockId) {
closeBlockInStream(mCachedPositionedReadStream);
mCachedPositionedReadStream = mBlockStore.getInStream(blockId, mOptions, mFailedWorkers);
mCachedPositionedReadStream = mBlockStore.getInStream(
blockInfo, mOptions, mFailedWorkers);
}
long offset = pos % mBlockSize;
int bytesRead = mCachedPositionedReadStream.positionedRead(offset, b, off,
Expand Down
7 changes: 7 additions & 0 deletions core/client/fs/src/test/java/alluxio/client/file/AlluxioFileInStreamTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -766,8 +766,15 @@ public void positionedReadRetry() throws Exception {
when(mBlockStore
.getInStream(eq(0L), any(InStreamOptions.class), any()))
.thenReturn(brokenStream).thenReturn(workingStream);
when(mBlockStore
.getInStream(eq(new BlockInfo().setBlockId(0)), any(InStreamOptions.class), any()))
.thenReturn(brokenStream).thenReturn(workingStream);
when(brokenStream.positionedRead(anyLong(), any(byte[].class), anyInt(), anyInt()))
.thenThrow(new UnavailableException("test exception"));
when(mBlockStore.getInfo(anyLong())).thenAnswer(invocation -> {
long blockId = invocation.getArgument(0);
return mStatus.getBlockInfo(blockId);
});

byte[] b = new byte[(int) BLOCK_LENGTH * 2];
mTestStream.positionedRead(BLOCK_LENGTH / 2, b, 0, b.length);
Expand Down