Skip to content

HADOOP-18546. ABFS:disable purging list of in progress reads in abfs stream closed #5176

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

Merged
steveloughran merged 16 commits into from
Dec 7, 2022
Merged

HADOOP-18546. ABFS:disable purging list of in progress reads in abfs stream closed #5176

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
54e706a
InProgressList not to be purged on the stream close.
saxenapranav Dec 1, 2022
39c3403
added test. concurrency issues
saxenapranav Dec 1, 2022
8cad3d5
fixed consistency issue in the testAbfsInputStream new tests
saxenapranav Dec 1, 2022
a8eb44b
improved javadocs
saxenapranav Dec 1, 2022
11ff043
Merge branch 'trunk' into HADOOP-18546
saxenapranav Dec 1, 2022
ddefd2e
added wait after test run to delete any completed buffers
saxenapranav Dec 1, 2022
6a1e0a6
test simplified.
saxenapranav Dec 1, 2022
02d39ca
Revert "HADOOP-18528. Disable abfs prefetching by default (#5134)"
saxenapranav Dec 1, 2022
2a6dcb6
true value for fs.azure.enable.readahead in core-default.xml; basic A…
saxenapranav Dec 2, 2022
fc833f2
refactored testStreamPurgeDuringReadAheadCallExecuting to remove depe…
saxenapranav Dec 2, 2022
11dd7bf
Revert "refactored testStreamPurgeDuringReadAheadCallExecuting to rem…
saxenapranav Dec 2, 2022
ac1e758
removal of assertion for inprogress to completedList and eviction whi…
saxenapranav Dec 2, 2022
64546a5
javadoc for test: refactor
saxenapranav Dec 2, 2022
4902598
checkstyle:magicNumber fix
saxenapranav Dec 5, 2022
c76f610
try-with-resources block for abfsInputStream in test; usage of String…
saxenapranav Dec 7, 2022
455a472
Merge branch 'trunk' into HADOOP-18546
saxenapranav Dec 7, 2022
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
7 changes: 0 additions & 7 deletions hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2166,13 +2166,6 @@ The switch to turn S3A auditing on or off.
<description>The AbstractFileSystem for gs: uris.</description>
</property>

<property>
<name>fs.azure.enable.readahead</name>
<value>false</value>
<description>Disable readahead/prefetching in AbfsInputStream.
See HADOOP-18521</description>
</property>

<property>
<name>io.seqfile.compress.blocksize</name>
<value>1000000</value>
Expand Down
2 changes: 1 addition & 1 deletion ...azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ public final class FileSystemConfigurations {
public static final boolean DEFAULT_ABFS_LATENCY_TRACK = false;
public static final long DEFAULT_SAS_TOKEN_RENEW_PERIOD_FOR_STREAMS_IN_SECONDS = 120;

public static final boolean DEFAULT_ENABLE_READAHEAD = false;
public static final boolean DEFAULT_ENABLE_READAHEAD = true;
public static final String DEFAULT_FS_AZURE_USER_AGENT_PREFIX = EMPTY_STRING;
public static final String DEFAULT_VALUE_UNKNOWN = "UNKNOWN";

Expand Down
2 changes: 1 addition & 1 deletion ...op-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStreamContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ public class AbfsInputStreamContext extends AbfsStreamContext {

private boolean tolerateOobAppends;

private boolean isReadAheadEnabled = false;
private boolean isReadAheadEnabled = true;

private boolean alwaysReadBufferSize;

Expand Down
1 change: 0 additions & 1 deletion .../hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -544,7 +544,6 @@ public synchronized void purgeBuffersForStream(AbfsInputStream stream) {
LOGGER.debug("Purging stale buffers for AbfsInputStream {} ", stream);
readAheadQueue.removeIf(readBuffer -> readBuffer.getStream() == stream);
purgeList(stream, completedReadList);
purgeList(stream, inProgressList);
}

/**
Expand Down
2 changes: 0 additions & 2 deletions ...src/test/java/org/apache/hadoop/fs/azurebfs/contract/ITestAbfsFileSystemContractSeek.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@

import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_AHEAD_RANGE;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_BUFFER_SIZE;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_READAHEAD;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
Expand Down Expand Up @@ -69,7 +68,6 @@ protected Configuration createConfiguration() {
protected AbstractFSContract createContract(final Configuration conf) {
conf.setInt(AZURE_READ_AHEAD_RANGE, MIN_BUFFER_SIZE);
conf.setInt(AZURE_READ_BUFFER_SIZE, MIN_BUFFER_SIZE);
conf.setBoolean(FS_AZURE_ENABLE_READAHEAD, true);
return new AbfsFileSystemContract(conf, isSecure);
}

Expand Down
113 changes: 111 additions & 2 deletions ...adoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsInputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,13 @@

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.ExecutionException;

import org.assertj.core.api.Assertions;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
Expand Down Expand Up @@ -82,6 +84,16 @@ public class TestAbfsInputStream extends
REDUCED_READ_BUFFER_AGE_THRESHOLD * 10; // 30 sec
private static final int ALWAYS_READ_BUFFER_SIZE_TEST_FILE_SIZE = 16 * ONE_MB;

@After
public void afterTest() throws InterruptedException {
Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

override teardown(), call superclass. that way you know the order of things happening

Copy link
Contributor Author

@saxenapranav saxenapranav Dec 2, 2022

Choose a reason for hiding this comment

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

Have taken it in new revision.

//thread wait so that previous test's inProgress buffers are processed and removed.
Thread.sleep(10000l);
Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

don't like this as it potentially ladds 10s to a test run, one which could still be a bit flaky.

what about using testResetReadBufferManager()?

Copy link
Contributor Author

@saxenapranav saxenapranav Dec 2, 2022

Choose a reason for hiding this comment

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

Have refactored in the new revision:

  1. Override of teardown()
  2. Usage of testResetReadBufferManager

ReadBufferManager readBufferManager = ReadBufferManager.getBufferManager();
while(readBufferManager.getCompletedReadListSize() > 0) {
readBufferManager.callTryEvict();
}
}

private AbfsRestOperation getMockRestOp() {
AbfsRestOperation op = mock(AbfsRestOperation.class);
AbfsHttpOperation httpOp = mock(AbfsHttpOperation.class);
Expand All @@ -106,7 +118,6 @@ private AbfsClient getMockAbfsClient() {
private AbfsInputStream getAbfsInputStream(AbfsClient mockAbfsClient,
String fileName) throws IOException {
AbfsInputStreamContext inputStreamContext = new AbfsInputStreamContext(-1);
inputStreamContext.isReadAheadEnabled(true);
// Create AbfsInputStream with the client instance
AbfsInputStream inputStream = new AbfsInputStream(
mockAbfsClient,
Expand All @@ -132,7 +143,6 @@ public AbfsInputStream getAbfsInputStream(AbfsClient abfsClient,
boolean alwaysReadBufferSize,
int readAheadBlockSize) throws IOException {
AbfsInputStreamContext inputStreamContext = new AbfsInputStreamContext(-1);
inputStreamContext.isReadAheadEnabled(true);
// Create AbfsInputStream with the client instance
AbfsInputStream inputStream = new AbfsInputStream(
abfsClient,
Expand Down Expand Up @@ -495,6 +505,105 @@ public void testSuccessfulReadAhead() throws Exception {
checkEvictedStatus(inputStream, 0, true);
}

/**
* This test expects InProgressList is not purged by the inputStream close.
* The readBuffer will move to completedList and then finally should get evicted.
*/
@Test
public void testStreamPurgeDuringReadAheadCallExecuting() throws Exception {
AbfsClient client = getMockAbfsClient();
AbfsRestOperation successOp = getMockRestOp();

Mockito.doAnswer(invocationOnMock -> {
//sleeping thread to mock the network latency from client to backend.
Thread.sleep(3000l);
return successOp;
})
.when(client)
Copy link
Contributor

@snvijaya snvijaya Dec 1, 2022

Choose a reason for hiding this comment

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

The test is trying to unit test a bigger scope of existing inprogress buffer moving to completed list. Will be nice to scope the test to inProgressList and freelist counts, before and after close.

At this client.read() mock, I would suggest mocks that will invoke a large sleep for each read. That way after queueReadAheads call and a 1 sec sleep, 3 buffers will be stuck inProgessList and the freeeList should show 13 free. The asserts should continue to hold to same numbers post close as well.

Copy link
Contributor Author

@saxenapranav saxenapranav Dec 1, 2022

Choose a reason for hiding this comment

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

Have taken the change of sleep and assertion on freeList also included in the tests.

Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

this is very brittle being timing based. normally I'd say "no" here, but I know I have a forthcoming pr which uses object.wait/notify to synchronize
https://github.com/apache/hadoop/pull/5117/files#diff-e829dbaa29faf05ae0b331439e9aec3cd02248464a097c86a0227783337b9b76R370

if this test causes problems it should do the same

Copy link
Contributor

@snvijaya snvijaya Dec 2, 2022

Choose a reason for hiding this comment

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

Hi Steve, the sleep time on these mock threads are meant to hold the thread blocked while the test goes ahead with asserts after queuing reads and asserts after close. The sleep of 1 second (which will block the main test thread) after queueing reads has been consistent with the timing expectations with pre-existing tests in this class doing the same, however I agree that this test has lot more going beyond the close which needs time synchronization, which can make the test brittle.

Hi Pranav, The test asserts post line 566 starting from 3 sec sleep are validations for correct movement of inprogress buffers to completed list and their evictions, which is a functionality that this PR change does not interfere. I would suggest that we take them out and evaluate if pre-existing test coverage doesnt handle it already. If there are gaps, lets take it in separate PR.

Copy link
Contributor Author

@saxenapranav saxenapranav Dec 2, 2022

Choose a reason for hiding this comment

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

Thanks. I have removed the assertion on inProgress to completedList and the eviction.

Copy link
Contributor

@steveloughran steveloughran Dec 2, 2022

Choose a reason for hiding this comment

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

it may be slow, but at least there's no assertion that something finishes before a specific timeout. those are the tests which really have problems on slow networks/overloaded systems

.read(any(String.class), any(Long.class), any(byte[].class),
any(Integer.class), any(Integer.class), any(String.class),
any(String.class), any(TracingContext.class));

AbfsInputStream inputStream = getAbfsInputStream(client,
"testSuccessfulReadAhead.txt");
queueReadAheads(inputStream);

final ReadBufferManager readBufferManager
= ReadBufferManager.getBufferManager();

//Sleeping to give ReadBufferWorker to pick the readBuffers for processing.
Thread.sleep(1000l);
Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

1_000L


Assertions.assertThat(
getStreamRelatedBufferCount(readBufferManager.getInProgressCopiedList(),
inputStream))
.describedAs("InProgressList should have 3 elements")
.isEqualTo(3);
Assertions.assertThat(readBufferManager.getFreeListCopy().size())
Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

use .hasSize(13) in the assert, so assertj will provide info about the list if there's a mismatch

.describedAs("FreeList should have 13 elements")
.isEqualTo(13);
Assertions.assertThat(readBufferManager.getCompletedReadListCopy().size())
.describedAs("CompletedList should have 0 elements")
.isEqualTo(0);

inputStream.close();
Copy link
Contributor

@steveloughran steveloughran Dec 6, 2022

Choose a reason for hiding this comment

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

the problem with the close() here is that it will only be reached if the assertions hold. if anything goes wrong, an exception is raised and the stream kept open, with whatever resources it consumes.

it should be closed in a finally block or the stream opened in a try-with-resources clause. thanks


Assertions.assertThat(
getStreamRelatedBufferCount(readBufferManager.getInProgressCopiedList(),
inputStream))
.describedAs("InProgressList should have 3 elements")
.isEqualTo(3);
Assertions.assertThat(getStreamRelatedBufferCount(
readBufferManager.getCompletedReadListCopy(), inputStream))
.describedAs("CompletedList should have 0 elements")
.isEqualTo(0);
Assertions.assertThat(readBufferManager.getFreeListCopy().size())
Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

use .hasSize(13)

.describedAs("FreeList should have 13 elements")
.isEqualTo(13);

//Sleep so that response from mockedClient gets back to ReadBufferWorker and
// can populate into completedList.
Thread.sleep(3000l);

Assertions.assertThat(getStreamRelatedBufferCount(
readBufferManager.getCompletedReadListCopy(), inputStream))
.describedAs("CompletedList should have 3 elements")
.isEqualTo(3);
Assertions.assertThat(readBufferManager.getFreeListCopy().size())
.describedAs("FreeList should have 13 elements")
.isEqualTo(13);
Assertions.assertThat(
getStreamRelatedBufferCount(readBufferManager.getInProgressCopiedList(),
inputStream))
.describedAs("InProgressList should have 0 elements")
.isEqualTo(0);

Thread.sleep(readBufferManager.getThresholdAgeMilliseconds());

readBufferManager.callTryEvict();
readBufferManager.callTryEvict();
readBufferManager.callTryEvict();

Assertions.assertThat(getStreamRelatedBufferCount(
readBufferManager.getCompletedReadListCopy(), inputStream))
.describedAs("CompletedList should have 0 elements")
.isEqualTo(0);
Assertions.assertThat(readBufferManager.getFreeListCopy().size())
Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

use .hasSize()

.describedAs("FreeList should have 16 elements")
.isEqualTo(16);
}

private int getStreamRelatedBufferCount(final List<ReadBuffer> bufferList,
final AbfsInputStream inputStream) {
int count = 0;
Copy link
Contributor

@steveloughran steveloughran Dec 1, 2022

Choose a reason for hiding this comment

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

prefer java8 streaming

bufferList.stream()
  .filter(buffer -> buffer.getStream() == inputStream)
  .count()

for (ReadBuffer buffer : bufferList) {
if (buffer.getStream() == inputStream) {
count++;
}
}
return count;
}

/**
* This test expects ReadAheadManager to throw exception if the read ahead
* thread had failed within the last thresholdAgeMilliseconds.
Expand Down