Merge pull request #4000 from jianx9/dev-v2

Provide an option to skip file descriptor sync in CacheDataSink. closeCurrentOutputStream
This commit is contained in:
ojw28 2018-04-16 11:07:11 +01:00 committed by GitHub
commit 650f96fd6e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 59 additions and 15 deletions

View File

@ -43,6 +43,7 @@ public final class CacheDataSink implements DataSink {
private final Cache cache; private final Cache cache;
private final long maxCacheFileSize; private final long maxCacheFileSize;
private final int bufferSize; private final int bufferSize;
private final boolean syncFileDescriptor;
private DataSpec dataSpec; private DataSpec dataSpec;
private File file; private File file;
@ -72,7 +73,20 @@ public final class CacheDataSink implements DataSink {
* multiple cache files. * multiple cache files.
*/ */
public CacheDataSink(Cache cache, long maxCacheFileSize) { public CacheDataSink(Cache cache, long maxCacheFileSize) {
this(cache, maxCacheFileSize, DEFAULT_BUFFER_SIZE); this(cache, maxCacheFileSize, DEFAULT_BUFFER_SIZE, true);
}
/**
* Constructs a CacheDataSink using the {@link #DEFAULT_BUFFER_SIZE}.
*
* @param cache The cache into which data should be written.
* @param maxCacheFileSize The maximum size of a cache file, in bytes. If the sink is opened for
* a {@link DataSpec} whose size exceeds this value, then the data will be fragmented into
* multiple cache files.
* @param syncFileDescriptor Skip file descriptor sync when closing current output stream.
*/
public CacheDataSink(Cache cache, long maxCacheFileSize, boolean syncFileDescriptor) {
this(cache, maxCacheFileSize, DEFAULT_BUFFER_SIZE, syncFileDescriptor);
} }
/** /**
@ -82,11 +96,13 @@ public final class CacheDataSink implements DataSink {
* multiple cache files. * multiple cache files.
* @param bufferSize The buffer size in bytes for writing to a cache file. A zero or negative * @param bufferSize The buffer size in bytes for writing to a cache file. A zero or negative
* value disables buffering. * value disables buffering.
* @param syncFileDescriptor Sync file descriptor when closing current output stream.
*/ */
public CacheDataSink(Cache cache, long maxCacheFileSize, int bufferSize) { public CacheDataSink(Cache cache, long maxCacheFileSize, int bufferSize, boolean syncFileDescriptor) {
this.cache = Assertions.checkNotNull(cache); this.cache = Assertions.checkNotNull(cache);
this.maxCacheFileSize = maxCacheFileSize; this.maxCacheFileSize = maxCacheFileSize;
this.bufferSize = bufferSize; this.bufferSize = bufferSize;
this.syncFileDescriptor = syncFileDescriptor;
} }
@Override @Override
@ -170,7 +186,9 @@ public final class CacheDataSink implements DataSink {
boolean success = false; boolean success = false;
try { try {
outputStream.flush(); outputStream.flush();
if (syncFileDescriptor) {
underlyingFileOutputStream.getFD().sync(); underlyingFileOutputStream.getFD().sync();
}
success = true; success = true;
} finally { } finally {
Util.closeQuietly(outputStream); Util.closeQuietly(outputStream);

View File

@ -44,7 +44,7 @@ public final class CacheDataSinkFactory implements DataSink.Factory {
@Override @Override
public DataSink createDataSink() { public DataSink createDataSink() {
return new CacheDataSink(cache, maxCacheFileSize, bufferSize); return new CacheDataSink(cache, maxCacheFileSize, bufferSize, true);
} }
} }

View File

@ -79,29 +79,49 @@ public final class CacheDataSourceTest {
@Test @Test
public void testCacheAndRead() throws Exception { public void testCacheAndRead() throws Exception {
assertCacheAndRead(false, false); assertCacheAndRead(false, false, true);
} }
@Test @Test
public void testCacheAndReadUnboundedRequest() throws Exception { public void testCacheAndReadUnboundedRequest() throws Exception {
assertCacheAndRead(true, false); assertCacheAndRead(true, false, true);
} }
@Test @Test
public void testCacheAndReadUnknownLength() throws Exception { public void testCacheAndReadUnknownLength() throws Exception {
assertCacheAndRead(false, true); assertCacheAndRead(false, true, true);
} }
@Test @Test
public void testCacheAndReadUnboundedRequestUnknownLength() throws Exception { public void testCacheAndReadUnboundedRequestUnknownLength() throws Exception {
assertCacheAndRead(true, true); assertCacheAndRead(true, true, true);
}
@Test
public void testCacheAndReadSkipFDSync() throws Exception {
assertCacheAndRead(false, false, false);
}
@Test
public void testCacheAndReadUnboundedRequestSkipFDSync() throws Exception {
assertCacheAndRead(true, false, false);
}
@Test
public void testCacheAndReadUnknownLengthSkipFDSync() throws Exception {
assertCacheAndRead(false, true, false);
}
@Test
public void testCacheAndReadUnboundedRequestUnknownLengthSkipFDSync() throws Exception {
assertCacheAndRead(true, true, false);
} }
@Test @Test
public void testUnsatisfiableRange() throws Exception { public void testUnsatisfiableRange() throws Exception {
// Bounded request but the content length is unknown. This forces all data to be cached but not // Bounded request but the content length is unknown. This forces all data to be cached but not
// the length // the length
assertCacheAndRead(false, true); assertCacheAndRead(false, true, true);
// Now do an unbounded request. This will read all of the data from cache and then try to read // Now do an unbounded request. This will read all of the data from cache and then try to read
// more from upstream which will cause to a 416 so CDS will store the length. // more from upstream which will cause to a 416 so CDS will store the length.
@ -347,10 +367,11 @@ public final class CacheDataSourceTest {
cacheDataSource.close(); cacheDataSource.close();
} }
private void assertCacheAndRead(boolean unboundedRequest, boolean simulateUnknownLength) private void assertCacheAndRead(boolean unboundedRequest, boolean simulateUnknownLength,
boolean syncFD)
throws IOException { throws IOException {
// Read all data from upstream and write to cache // Read all data from upstream and write to cache
CacheDataSource cacheDataSource = createCacheDataSource(false, simulateUnknownLength); CacheDataSource cacheDataSource = createCacheDataSource(false, simulateUnknownLength, syncFD);
assertReadDataContentLength(cacheDataSource, unboundedRequest, simulateUnknownLength); assertReadDataContentLength(cacheDataSource, unboundedRequest, simulateUnknownLength);
// Just read from cache // Just read from cache
@ -391,14 +412,19 @@ public final class CacheDataSourceTest {
private CacheDataSource createCacheDataSource(boolean setReadException, private CacheDataSource createCacheDataSource(boolean setReadException,
boolean simulateUnknownLength) { boolean simulateUnknownLength) {
return createCacheDataSource(setReadException, simulateUnknownLength, return createCacheDataSource(setReadException, simulateUnknownLength, true);
CacheDataSource.FLAG_BLOCK_ON_CACHE);
} }
private CacheDataSource createCacheDataSource(boolean setReadException, private CacheDataSource createCacheDataSource(boolean setReadException,
boolean simulateUnknownLength, @CacheDataSource.Flags int flags) { boolean simulateUnknownLength, boolean syncFD) {
return createCacheDataSource(setReadException, simulateUnknownLength,
CacheDataSource.FLAG_BLOCK_ON_CACHE, syncFD);
}
private CacheDataSource createCacheDataSource(boolean setReadException,
boolean simulateUnknownLength, @CacheDataSource.Flags int flags, boolean syncFD) {
return createCacheDataSource(setReadException, simulateUnknownLength, flags, return createCacheDataSource(setReadException, simulateUnknownLength, flags,
new CacheDataSink(cache, MAX_CACHE_FILE_SIZE)); new CacheDataSink(cache, MAX_CACHE_FILE_SIZE, syncFD));
} }
private CacheDataSource createCacheDataSource(boolean setReadException, private CacheDataSource createCacheDataSource(boolean setReadException,