Skip to content

Storage: Fix slow download performance #5791

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
ajaaym merged 8 commits into from
Jul 31, 2019
Merged

Storage: Fix slow download performance #5791

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c00cb4c
download blob using direct download method
ajaaym Jul 19, 2019
615dab7
update test case
ajaaym Jul 19, 2019
ae1ffab
fix format
ajaaym Jul 19, 2019
1bdc864
add test case for retries
ajaaym Jul 19, 2019
48777c2
remove method
ajaaym Jul 30, 2019
0aebf4c
update test retry setting
ajaaym Jul 30, 2019
58469f5
fix lint and add comment
ajaaym Jul 31, 2019
cb65519
fix lint
ajaaym Jul 31, 2019
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
53 changes: 41 additions & 12 deletions google-cloud-clients/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@

package com.google.cloud.storage;

import static com.google.cloud.RetryHelper.runWithRetries;
import static com.google.cloud.storage.Blob.BlobSourceOption.toGetOptions;
import static com.google.cloud.storage.Blob.BlobSourceOption.toSourceOptions;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.concurrent.Executors.callable;

import com.google.api.services.storage.model.StorageObject;
import com.google.auth.ServiceAccountSigner;
Expand All @@ -34,13 +36,11 @@
import com.google.cloud.storage.spi.v1.StorageRpc;
import com.google.common.base.Function;
import com.google.common.io.BaseEncoding;
import com.google.common.io.CountingOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.OutputStream;
import java.net.URL;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.Key;
Expand Down Expand Up @@ -211,20 +211,49 @@ static Storage.BlobGetOption[] toGetOptions(BlobInfo blobInfo, BlobSourceOption.
* @throws StorageException upon failure
*/
public void downloadTo(Path path, BlobSourceOption... options) {
try (OutputStream outputStream = Files.newOutputStream(path);
ReadChannel reader = reader(options)) {
WritableByteChannel channel = Channels.newChannel(outputStream);
ByteBuffer bytes = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
while (reader.read(bytes) > 0) {
bytes.flip();
channel.write(bytes);
bytes.clear();
}
try (OutputStream outputStream = Files.newOutputStream(path)) {
downloadTo(outputStream, options);
} catch (IOException e) {
throw new StorageException(e);
}
}

/**
* Downloads this blob to the given output stream using specified blob read options.
*
* @param outputStream
* @param options
*/
public void downloadTo(OutputStream outputStream, BlobSourceOption... options) {
final CountingOutputStream countingOutputStream = new CountingOutputStream(outputStream);
final StorageRpc storageRpc = this.options.getStorageRpcV1();
final Map<StorageRpc.Option, ?> requestOptions = StorageImpl.optionMap(getBlobId(), options);
runWithRetries(
callable(
new Runnable() {
@Override
public void run() {
storageRpc.read(
getBlobId().toPb(),
requestOptions,
countingOutputStream.getCount(),
countingOutputStream);
}
}),
this.options.getRetrySettings(),
StorageImpl.EXCEPTION_HANDLER,
this.options.getClock());
}

/**
* Downloads this blob to the given output stream.
*
* @param outputStream
*/
public void downloadTo(OutputStream outputStream) {
downloadTo(outputStream, new BlobSourceOption[0]);
Copy link
Contributor

@frankyn frankyn Jul 25, 2019

Choose a reason for hiding this comment

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

This overloaded method doesn't look necessary. My varargs are rusty but if no Options are specified in downloadTo(OutputStream os, Options..) it should still work.

Copy link
Contributor Author

@ajaaym ajaaym Jul 30, 2019

Choose a reason for hiding this comment

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

yes its not needed. removed it.

}

/**
* Downloads this blob to the given file path.
*
Expand Down
106 changes: 90 additions & 16 deletions ...e-cloud-clients/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,9 @@
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;

import com.google.api.core.ApiClock;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.services.storage.model.StorageObject;
import com.google.cloud.ReadChannel;
import com.google.cloud.storage.Acl.Project;
import com.google.cloud.storage.Acl.Project.ProjectRole;
Expand All @@ -41,12 +44,13 @@
import com.google.cloud.storage.Blob.BlobSourceOption;
import com.google.cloud.storage.Storage.BlobWriteOption;
import com.google.cloud.storage.Storage.CopyRequest;
import com.google.cloud.storage.spi.v1.StorageRpc;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.BaseEncoding;
import java.io.File;
import java.io.OutputStream;
import java.net.URL;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.security.Key;
import java.util.List;
Expand All @@ -58,6 +62,7 @@
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.threeten.bp.Duration;
Copy link
Contributor

@frankyn frankyn Jul 30, 2019

Choose a reason for hiding this comment

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

Is this a necessary import. It looks unused.


public class BlobTest {

Expand Down Expand Up @@ -130,6 +135,28 @@ public class BlobTest {
private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
private static final Key KEY =
new SecretKeySpec(BaseEncoding.base64().decode(BASE64_KEY), "AES256");
private static final RetrySettings RETRY_SETTINGS =
Copy link
Contributor

@frankyn frankyn Jul 25, 2019

Choose a reason for hiding this comment

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

There's heavy retry settings in the tests. Is this by design and how do they help?

Copy link
Contributor Author

@ajaaym ajaaym Jul 30, 2019

Choose a reason for hiding this comment

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

updated retry settings needed for test.

RetrySettings.newBuilder()
.setInitialRetryDelay(Duration.ofMillis(100L))
.setRetryDelayMultiplier(1.3)
.setMaxRetryDelay(Duration.ofMillis(60000L))
.setInitialRpcTimeout(Duration.ofMillis(20000L))
.setRpcTimeoutMultiplier(1.0)
.setMaxRpcTimeout(Duration.ofMillis(20000L))
.setTotalTimeout(Duration.ofMillis(600000L))
.build();
private static final ApiClock API_CLOCK =
Copy link
Contributor

@frankyn frankyn Jul 30, 2019

Choose a reason for hiding this comment

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

Please document how retries are being exercised in these unit tests. It will help for posterity.

new ApiClock() {
@Override
public long nanoTime() {
return 42_000_000_000L;
}

@Override
public long millisTime() {
return 42_000L;
}
};

private Storage storage;
private Blob blob;
Expand Down Expand Up @@ -566,28 +593,75 @@ public void testBuilder() {
@Test
public void testDownload() throws Exception {
final byte[] expected = {1, 2};
StorageRpc mockStorageRpc = createNiceMock(StorageRpc.class);
expect(storage.getOptions()).andReturn(mockOptions).times(1);
replay(storage);
expect(mockOptions.getStorageRpcV1()).andReturn(mockStorageRpc);
expect(mockOptions.getRetrySettings()).andReturn(RETRY_SETTINGS);
expect(mockOptions.getClock()).andReturn(API_CLOCK);
replay(mockOptions);
blob = new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO));
expect(
mockStorageRpc.read(
anyObject(StorageObject.class),
anyObject(Map.class),
eq(0l),
anyObject(OutputStream.class)))
.andAnswer(
new IAnswer<Long>() {
@Override
public Long answer() throws Throwable {
((OutputStream) getCurrentArguments()[3]).write(expected);
return 2l;
}
});
replay(mockStorageRpc);
File file = File.createTempFile("blob", ".tmp");
blob.downloadTo(file.toPath());
byte actual[] = Files.readAllBytes(file.toPath());
assertArrayEquals(expected, actual);
}

initializeExpectedBlob(2);
ReadChannel channel = createNiceMock(ReadChannel.class);
@Test
public void testDownloadWithRetries() throws Exception {
final byte[] expected = {1, 2};
StorageRpc mockStorageRpc = createNiceMock(StorageRpc.class);
expect(storage.getOptions()).andReturn(mockOptions);
expect(storage.reader(BLOB_INFO.getBlobId())).andReturn(channel);
replay(storage);
// First read should return 2 bytes.
expect(channel.read(anyObject(ByteBuffer.class)))
expect(mockOptions.getStorageRpcV1()).andReturn(mockStorageRpc);
expect(mockOptions.getRetrySettings()).andReturn(RETRY_SETTINGS);
expect(mockOptions.getClock()).andReturn(API_CLOCK);
replay(mockOptions);
blob = new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO));
expect(
mockStorageRpc.read(
anyObject(StorageObject.class),
anyObject(Map.class),
eq(0l),
anyObject(OutputStream.class)))
.andAnswer(
new IAnswer<Integer>() {
new IAnswer<Long>() {
@Override
public Integer answer() throws Throwable {
// Modify the argument to match the expected behavior of `read`.
((ByteBuffer) getCurrentArguments()[0]).put(expected);
return 2;
public Long answer() throws Throwable {
((OutputStream) getCurrentArguments()[3]).write(expected[0]);
throw new StorageException(504, "error");
}
});
// Second read should return 0 bytes.
expect(channel.read(anyObject(ByteBuffer.class))).andReturn(0);
replay(channel);
initializeBlob();

expect(
mockStorageRpc.read(
anyObject(StorageObject.class),
anyObject(Map.class),
eq(1l),
anyObject(OutputStream.class)))
.andAnswer(
new IAnswer<Long>() {
@Override
public Long answer() throws Throwable {
((OutputStream) getCurrentArguments()[3]).write(expected[1]);
return 1l;
}
});
replay(mockStorageRpc);
File file = File.createTempFile("blob", ".tmp");
blob.downloadTo(file.toPath());
byte actual[] = Files.readAllBytes(file.toPath());
Expand Down