Skip to content

Commit 702ab2b

Browse files
BenWhiteheadBenWhitehead
authored
feat: make it possible to disable the buffer of ReadChannels returned from Storage.reader (#1974)
For some scenarios, an external client needs the ability to manage buffer blocking itself. To support this, providing 0 to ReadChannel#setChunkSize will disable buffering allowing for client to avoid the need to alight multiple levels of buffer alignments. Because the buffering is disabled, that means reads can be much more variable in size and individually more impacted by small network latencies (with buffering these can be amortized by the buffer making followup read faster). This is considered advanced usage and will require more work from the client integrator.
1 parent f1b9493 commit 702ab2b

6 files changed

Lines changed: 113 additions & 42 deletions

File tree

‎google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java‎

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -23,12 +23,12 @@
2323
import com.google.api.core.ApiFutureCallback;
2424
import com.google.api.core.ApiFutures;
2525
import com.google.api.core.SettableApiFuture;
26-
import com.google.cloud.storage.BufferedReadableByteChannelSession.BufferedReadableByteChannel;
2726
import com.google.cloud.storage.Conversions.Decoder;
2827
import com.google.common.util.concurrent.MoreExecutors;
2928
import java.io.IOException;
3029
import java.nio.ByteBuffer;
3130
import java.nio.channels.ClosedChannelException;
31+
import java.nio.channels.ReadableByteChannel;
3232
import org.checkerframework.checker.nullness.qual.Nullable;
3333

3434
abstract class BaseStorageReadChannel<T> implements StorageReadChannel {
@@ -40,7 +40,7 @@ abstract class BaseStorageReadChannel<T> implements StorageReadChannel {
4040
private ByteRangeSpec byteRangeSpec;
4141
private int chunkSize = _2MiB;
4242
private BufferHandle bufferHandle;
43-
private LazyReadChannel<T> lazyReadChannel;
43+
private LazyReadChannel<?, T> lazyReadChannel;
4444

4545
protected BaseStorageReadChannel(Decoder<T, BlobInfo> objectDecoder) {
4646
this.objectDecoder = objectDecoder;
@@ -64,15 +64,18 @@ public final synchronized boolean isOpen() {
6464
public final synchronized void close() {
6565
open = false;
6666
if (internalGetLazyChannel().isOpen()) {
67-
StorageException.wrapIOException(internalGetLazyChannel().getChannel()::close);
67+
ReadableByteChannel channel = internalGetLazyChannel().getChannel();
68+
StorageException.wrapIOException(channel::close);
6869
}
6970
}
7071

7172
@Override
7273
public final synchronized StorageReadChannel setByteRangeSpec(ByteRangeSpec byteRangeSpec) {
7374
requireNonNull(byteRangeSpec, "byteRangeSpec must be non null");
74-
StorageException.wrapIOException(() -> maybeResetChannel(false));
75-
this.byteRangeSpec = byteRangeSpec;
75+
if (!this.byteRangeSpec.equals(byteRangeSpec)) {
76+
StorageException.wrapIOException(() -> maybeResetChannel(false));
77+
this.byteRangeSpec = byteRangeSpec;
78+
}
7679
return this;
7780
}
7881

@@ -95,7 +98,7 @@ public final synchronized int read(ByteBuffer dst) throws IOException {
9598
}
9699
try {
97100
// trap if the fact that tmp is already closed, and instead return -1
98-
BufferedReadableByteChannel tmp = internalGetLazyChannel().getChannel();
101+
ReadableByteChannel tmp = internalGetLazyChannel().getChannel();
99102
if (!tmp.isOpen()) {
100103
return -1;
101104
}
@@ -146,7 +149,7 @@ protected final T getResolvedObject() {
146149
}
147150
}
148151

149-
protected abstract LazyReadChannel<T> newLazyReadChannel();
152+
protected abstract LazyReadChannel<?, T> newLazyReadChannel();
150153

151154
private void maybeResetChannel(boolean freeBuffer) throws IOException {
152155
if (lazyReadChannel != null) {
@@ -162,9 +165,9 @@ private void maybeResetChannel(boolean freeBuffer) throws IOException {
162165
}
163166
}
164167

165-
private LazyReadChannel<T> internalGetLazyChannel() {
168+
private LazyReadChannel<?, T> internalGetLazyChannel() {
166169
if (lazyReadChannel == null) {
167-
LazyReadChannel<T> tmp = newLazyReadChannel();
170+
LazyReadChannel<?, T> tmp = newLazyReadChannel();
168171
ApiFuture<T> future = tmp.getSession().getResult();
169172
ApiFutures.addCallback(
170173
future,

‎google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java‎

Lines changed: 19 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.google.cloud.ReadChannel;
2222
import com.google.cloud.RestorableState;
2323
import com.google.cloud.storage.ApiaryUnbufferedReadableByteChannel.ApiaryReadRequest;
24+
import com.google.cloud.storage.HttpDownloadSessionBuilder.ReadableByteChannelSessionBuilder;
2425
import com.google.cloud.storage.spi.v1.StorageRpc;
2526
import com.google.common.base.MoreObjects;
2627
import java.io.Serializable;
@@ -56,16 +57,25 @@ public synchronized RestorableState<ReadChannel> capture() {
5657
apiaryReadRequest, blobReadChannelContext.getStorageOptions(), getChunkSize());
5758
}
5859

59-
protected LazyReadChannel<StorageObject> newLazyReadChannel() {
60+
protected LazyReadChannel<?, StorageObject> newLazyReadChannel() {
6061
return new LazyReadChannel<>(
61-
() ->
62-
ResumableMedia.http()
63-
.read()
64-
.byteChannel(blobReadChannelContext)
65-
.setAutoGzipDecompression(autoGzipDecompression)
66-
.buffered(getBufferHandle())
67-
.setApiaryReadRequest(getApiaryReadRequest())
68-
.build());
62+
() -> {
63+
ReadableByteChannelSessionBuilder b =
64+
ResumableMedia.http()
65+
.read()
66+
.byteChannel(blobReadChannelContext)
67+
.setAutoGzipDecompression(autoGzipDecompression);
68+
BufferHandle bufferHandle = getBufferHandle();
69+
// because we're erasing the specific type of channel, we need to declare it here.
70+
// If we don't, the compiler complains we're not returning a compliant type.
71+
ReadableByteChannelSession<?, StorageObject> session;
72+
if (bufferHandle.capacity() > 0) {
73+
session = b.buffered(bufferHandle).setApiaryReadRequest(getApiaryReadRequest()).build();
74+
} else {
75+
session = b.unbuffered().setApiaryReadRequest(getApiaryReadRequest()).build();
76+
}
77+
return session;
78+
});
6979
}
7080

7181
private ApiaryReadRequest getApiaryReadRequest() {

‎google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java‎

Lines changed: 21 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import com.google.api.gax.rpc.ServerStreamingCallable;
2020
import com.google.cloud.ReadChannel;
2121
import com.google.cloud.RestorableState;
22+
import com.google.cloud.storage.GapicDownloadSessionBuilder.ReadableByteChannelSessionBuilder;
2223
import com.google.storage.v2.Object;
2324
import com.google.storage.v2.ReadObjectRequest;
2425
import com.google.storage.v2.ReadObjectResponse;
@@ -46,17 +47,27 @@ public RestorableState<ReadChannel> capture() {
4647
}
4748

4849
@Override
49-
protected LazyReadChannel<Object> newLazyReadChannel() {
50+
protected LazyReadChannel<?, Object> newLazyReadChannel() {
5051
return new LazyReadChannel<>(
51-
() ->
52-
ResumableMedia.gapic()
53-
.read()
54-
.byteChannel(read)
55-
.setHasher(Hasher.noop())
56-
.setAutoGzipDecompression(autoGzipDecompression)
57-
.buffered(getBufferHandle())
58-
.setReadObjectRequest(getReadObjectRequest())
59-
.build());
52+
() -> {
53+
ReadableByteChannelSessionBuilder b =
54+
ResumableMedia.gapic()
55+
.read()
56+
.byteChannel(read)
57+
.setHasher(Hasher.noop())
58+
.setAutoGzipDecompression(autoGzipDecompression);
59+
BufferHandle bufferHandle = getBufferHandle();
60+
// because we're erasing the specific type of channel, we need to declare it here.
61+
// If we don't, the compiler complains we're not returning a compliant type.
62+
ReadableByteChannelSession<?, Object> session;
63+
if (bufferHandle.capacity() > 0) {
64+
session =
65+
b.buffered(getBufferHandle()).setReadObjectRequest(getReadObjectRequest()).build();
66+
} else {
67+
session = b.unbuffered().setReadObjectRequest(getReadObjectRequest()).build();
68+
}
69+
return session;
70+
});
6071
}
6172

6273
@NonNull

‎google-cloud-storage/src/main/java/com/google/cloud/storage/LazyReadChannel.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,26 +16,26 @@
1616

1717
package com.google.cloud.storage;
1818

19-
import com.google.cloud.storage.BufferedReadableByteChannelSession.BufferedReadableByteChannel;
19+
import java.nio.channels.ReadableByteChannel;
2020
import java.util.function.Supplier;
2121
import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
2222
import org.checkerframework.checker.nullness.qual.NonNull;
2323

24-
final class LazyReadChannel<T> {
24+
final class LazyReadChannel<RBC extends ReadableByteChannel, T> {
2525

26-
private final Supplier<BufferedReadableByteChannelSession<T>> sessionSupplier;
26+
private final Supplier<ReadableByteChannelSession<RBC, T>> sessionSupplier;
2727

28-
@MonotonicNonNull private volatile BufferedReadableByteChannelSession<T> session;
29-
@MonotonicNonNull private volatile BufferedReadableByteChannel channel;
28+
@MonotonicNonNull private volatile ReadableByteChannelSession<RBC, T> session;
29+
@MonotonicNonNull private volatile RBC channel;
3030

3131
private boolean open = false;
3232

33-
LazyReadChannel(Supplier<BufferedReadableByteChannelSession<T>> sessionSupplier) {
33+
LazyReadChannel(Supplier<ReadableByteChannelSession<RBC, T>> sessionSupplier) {
3434
this.sessionSupplier = sessionSupplier;
3535
}
3636

3737
@NonNull
38-
BufferedReadableByteChannel getChannel() {
38+
RBC getChannel() {
3939
if (channel != null) {
4040
return channel;
4141
} else {
@@ -50,7 +50,7 @@ BufferedReadableByteChannel getChannel() {
5050
}
5151

5252
@NonNull
53-
BufferedReadableByteChannelSession<T> getSession() {
53+
ReadableByteChannelSession<RBC, T> getSession() {
5454
if (session != null) {
5555
return session;
5656
} else {

‎google-cloud-storage/src/test/java/com/google/cloud/storage/LazyReadChannelTest.java‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -34,16 +34,18 @@ public final class LazyReadChannelTest {
3434

3535
@Test
3636
public void repeatedCallsOfGetSessionMustReturnTheSameInstance() {
37-
LazyReadChannel<String> lrc = new LazyReadChannel<>(this::newTestSession);
37+
LazyReadChannel<BufferedReadableByteChannel, String> lrc =
38+
new LazyReadChannel<>(this::newTestSession);
3839

39-
BufferedReadableByteChannelSession<String> session1 = lrc.getSession();
40-
BufferedReadableByteChannelSession<String> session2 = lrc.getSession();
40+
ReadableByteChannelSession<BufferedReadableByteChannel, String> session1 = lrc.getSession();
41+
ReadableByteChannelSession<BufferedReadableByteChannel, String> session2 = lrc.getSession();
4142
assertThat(session1).isSameInstanceAs(session2);
4243
}
4344

4445
@Test
4546
public void repeatedCallsOfGetChannelMustReturnTheSameInstance() {
46-
LazyReadChannel<String> lrc = new LazyReadChannel<>(this::newTestSession);
47+
LazyReadChannel<BufferedReadableByteChannel, String> lrc =
48+
new LazyReadChannel<>(this::newTestSession);
4749

4850
BufferedReadableByteChannel channel1 = lrc.getChannel();
4951
BufferedReadableByteChannel channel2 = lrc.getChannel();
@@ -52,7 +54,8 @@ public void repeatedCallsOfGetChannelMustReturnTheSameInstance() {
5254

5355
@Test
5456
public void isNotOpenUntilGetChannelIsCalled() {
55-
LazyReadChannel<String> lrc = new LazyReadChannel<>(this::newTestSession);
57+
LazyReadChannel<BufferedReadableByteChannel, String> lrc =
58+
new LazyReadChannel<>(this::newTestSession);
5659

5760
assertThat(lrc.isOpen()).isFalse();
5861
BufferedReadableByteChannel channel = lrc.getChannel();
@@ -63,7 +66,8 @@ public void isNotOpenUntilGetChannelIsCalled() {
6366

6467
@Test
6568
public void closingUnderlyingChannelClosesTheLazyReadChannel() throws IOException {
66-
LazyReadChannel<String> lrc = new LazyReadChannel<>(this::newTestSession);
69+
LazyReadChannel<BufferedReadableByteChannel, String> lrc =
70+
new LazyReadChannel<>(this::newTestSession);
6771

6872
BufferedReadableByteChannel channel = lrc.getChannel();
6973
assertThat(channel.isOpen()).isTrue();

‎google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageReadChannelTest.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import com.google.cloud.storage.it.runner.registry.Generator;
4343
import java.io.IOException;
4444
import java.nio.ByteBuffer;
45+
import java.nio.channels.ReadableByteChannel;
4546
import java.util.concurrent.ExecutionException;
4647
import java.util.concurrent.TimeUnit;
4748
import java.util.function.Function;
@@ -107,6 +108,35 @@ public void storageReadChannel_getObject_returns() throws Exception {
107108
}
108109
}
109110

111+
@Test
112+
// @CrossRun.Exclude(transports = Transport.GRPC)
113+
public void storageReadChannel_shouldAllowDisablingBufferingBySettingChunkSize_lteq0()
114+
throws IOException {
115+
int _512KiB = 512 * 1024;
116+
int _1MiB = 1024 * 1024;
117+
118+
final BlobInfo info;
119+
byte[] uncompressedBytes = DataGenerator.base64Characters().genBytes(_512KiB);
120+
{
121+
BlobInfo tmp = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
122+
Blob gen1 = storage.create(tmp, uncompressedBytes, BlobTargetOption.doesNotExist());
123+
info = gen1.asBlobInfo();
124+
}
125+
126+
try (ReadChannel c = storage.reader(info.getBlobId())) {
127+
c.setChunkSize(0);
128+
129+
ByteBuffer buf = ByteBuffer.allocate(_1MiB);
130+
// Because this is unbuffered, the underlying channel will not necessarily fill up the buf
131+
// in a single read call. Repeatedly read until full or EOF.
132+
int read = fillFrom(buf, c);
133+
assertThat(read).isEqualTo(_512KiB);
134+
String actual = xxd(buf);
135+
String expected = xxd(uncompressedBytes);
136+
assertThat(actual).isEqualTo(expected);
137+
}
138+
}
139+
110140
@Test
111141
public void storageReadChannel_getObject_404() {
112142
BlobId id = BlobId.of(bucket.getName(), generator.randomObjectName());
@@ -129,4 +159,17 @@ private static <T, F> void equalForField(T actual, T expected, Function<T, F> f)
129159
F eF = f.apply(expected);
130160
assertThat(aF).isEqualTo(eF);
131161
}
162+
163+
static int fillFrom(ByteBuffer buf, ReadableByteChannel c) throws IOException {
164+
int total = 0;
165+
while (buf.hasRemaining()) {
166+
int read = c.read(buf);
167+
if (read != -1) {
168+
total += read;
169+
} else {
170+
break;
171+
}
172+
}
173+
return total;
174+
}
132175
}

0 commit comments

Comments
 (0)