From e9de87e423638911b47efcc8d3758b98a6c02670 Mon Sep 17 00:00:00 2001 From: Dhriti Chopra Date: Tue, 27 Oct 2026 20:14:52 +0000 Subject: [PATCH] feat(storage): the session gives a fast open read to the matching readAs Second step of fast open BiDi reads (go/fast-open-bidi-read, b/552263062). ObjectReadSessionImpl keeps the fast open reads sent with the open, keyed by the config each was created from. readAs removes the read for an equal config and returns it, so the bytes that came back with the open are used and no second request is sent; any other readAs sends a new read exactly as before. Closing the session first closes fast open reads that were never asked for. An unclaimed asChannel() read fills its small buffer and blocks the stream's thread, so it must be closed (and its buffer emptied) before the stream, or close() would wait forever. Only one fast open read is supported for now; readSession rejects more than one before anything is sent. --- .../BaseObjectReadSessionStreamRead.java | 23 ++- .../cloud/storage/ObjectReadSessionImpl.java | 25 ++- .../cloud/storage/StorageDataClient.java | 14 +- .../cloud/storage/ITFastOpenFakeTest.java | 173 ++++++++++++++++++ 4 files changed, 227 insertions(+), 8 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseObjectReadSessionStreamRead.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseObjectReadSessionStreamRead.java index a590f404e57b..d868f09b9f4b 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseObjectReadSessionStreamRead.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseObjectReadSessionStreamRead.java @@ -56,7 +56,7 @@ abstract class BaseObjectReadSessionStreamRead protected final RangeSpec rangeSpec; protected final RetryContext retryContext; protected final AtomicLong readOffset; - protected boolean closed; + protected volatile boolean closed; protected boolean tombstoned; protected IOAutoCloseable onCloseCallback; @@ -328,8 +328,10 @@ public ApiFuture fail(Throwable t) { try { offer(new SmuggledFailure(t)); failFuture.set(null); - } catch (InterruptedIOException e) { - Thread.currentThread().interrupt(); + } catch (IOException e) { + if (e instanceof InterruptedIOException) { + Thread.currentThread().interrupt(); + } failFuture.setException(e); } return failFuture; @@ -433,9 +435,13 @@ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException { return read; } - private void offer(Closeable offer) throws InterruptedIOException { + private void offer(Closeable offer) throws IOException { try { queue.put(offer); + if (closed) { + // the reader closed while we waited, nobody will take this anymore + drainQueue(); + } } catch (InterruptedException e) { Thread.currentThread().interrupt(); InterruptedIOException ioe = new InterruptedIOException(); @@ -449,7 +455,14 @@ private void internalCleanup() throws IOException { if (leftovers != null) { leftovers.ref.close(); } - GrpcUtils.closeAll(queue); + drainQueue(); + } + + /** Remove and close everything in the queue, freeing space for a producer waiting to add. */ + private void drainQueue() throws IOException { + List drained = new ArrayList<>(); + queue.drainTo(drained); + GrpcUtils.closeAll(drained); } /** diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java index 2998f4d8ff33..085b63a14196 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java @@ -34,7 +34,10 @@ import java.util.Iterator; import java.util.List; import java.util.Locale; +import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.locks.ReentrantLock; @@ -52,6 +55,12 @@ final class ObjectReadSessionImpl implements ObjectReadSession { private final ConcurrentIdentityMap children; + // Fast open reads: reads sent in the request that opened the stream, keyed by the config each was + // created from. The first readAs with an equal config is given the read instead of sending a new + // one. + private final ConcurrentMap, ObjectReadSessionStreamRead> + fastOpenReads; + private volatile boolean open; ObjectReadSessionImpl( @@ -59,7 +68,8 @@ final class ObjectReadSessionImpl implements ObjectReadSession { ZeroCopyBidiStreamingCallable callable, ObjectReadSessionStream stream, ObjectReadSessionState state, - RetryContextProvider retryContextProvider) { + RetryContextProvider retryContextProvider, + Map, ObjectReadSessionStreamRead> fastOpenReads) { this.executor = executor; this.callable = callable; this.stream = stream; @@ -67,6 +77,7 @@ final class ObjectReadSessionImpl implements ObjectReadSession { this.resource = state.getMetadata(); this.retryContextProvider = retryContextProvider; this.children = new ConcurrentIdentityMap<>(); + this.fastOpenReads = new ConcurrentHashMap<>(fastOpenReads); this.open = true; } @@ -80,6 +91,14 @@ public Projection readAs(ReadProjectionConfig config) { checkState(open, "Session already closed"); switch (config.getType()) { case STREAM_READ: + ObjectReadSessionStreamRead fastOpenRead = fastOpenReads.remove(config); + if (fastOpenRead != null) { + // configs are final classes with a class-exact equals, so an equal config has the same + // Projection type + @SuppressWarnings("unchecked") + Projection projection = (Projection) fastOpenRead.project(); + return projection; + } long readId = state.newReadId(); ObjectReadSessionStreamRead read = config.cast().newRead(readId, retryContextProvider.create()); @@ -104,6 +123,10 @@ public void close() throws IOException { return; } open = false; + // close fast open reads nobody asked for before the stream: an unread channel read holds the + // stream's thread, and a completed read is out of the state, so stream.close() can't reach it + GrpcUtils.closeAll(fastOpenReads.values()); + fastOpenReads.clear(); List> closing = children.drainEntries((subStream, subStreamState) -> subStream.closeAsync()); stream.close(); diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java index 0ef051bb8c4a..cd3570a3a679 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java @@ -27,11 +27,14 @@ import com.google.cloud.storage.ReadProjectionConfig.ProjectionType; import com.google.cloud.storage.RetryContext.RetryContextProvider; import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import com.google.storage.v2.BidiReadObjectRequest; import com.google.storage.v2.BidiReadObjectResponse; import java.io.IOException; import java.time.Duration; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -66,6 +69,9 @@ ApiFuture readSession(BidiReadObjectRequest req, GrpcCallCont * Open a new {@link ObjectReadSession} using fast open: the ranges in {@code fastOpenConfigs} are * sent in the same request that opens the session, saving a round trip before the first bytes. * + *

The first {@code readAs} with a config equal to one of these is given that fast open read + * instead of sending a new one. + * * @param fastOpenConfigs the reads to send with the open, for example {@code * ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(0, 1024))}; empty to only * open the session @@ -77,6 +83,7 @@ ApiFuture readSession( checkArgument( req.getReadRangesList().isEmpty(), "ranges included in the initial request are not supported"); + checkArgument(fastOpenConfigs.size() <= 1, "at most one fast open read is supported"); ObjectReadSessionState state = new ObjectReadSessionState(ctx, req); ZeroCopyBidiStreamingCallable callable = @@ -86,18 +93,21 @@ ApiFuture readSession( // Add each fast open read to the session state. The open request below is built from the // state, so these reads are sent as part of the open. + Map, ObjectReadSessionStreamRead> fastOpenReads = new HashMap<>(); for (ReadProjectionConfig config : fastOpenConfigs) { long readId = state.newReadId(); ObjectReadSessionStreamRead read = config.cast().newRead(readId, retryContextProvider.create()); state.putOutstandingRead(readId, read); + fastOpenReads.put(config, read); } ApiFuture objectReadSessionFuture = ApiFutures.transform( stream, nowOpen -> - new ObjectReadSessionImpl(executor, callable, stream, state, retryContextProvider), + new ObjectReadSessionImpl( + executor, callable, stream, state, retryContextProvider, fastOpenReads), executor); stream.send(state.getOpenArguments().getReq()); return objectReadSessionFuture; @@ -132,7 +142,7 @@ ApiFuture> fastOpenReadSessio nowOpen -> new FastOpenObjectReadSession<>( new ObjectReadSessionImpl( - executor, callable, stream, state, retryContextProvider), + executor, callable, stream, state, retryContextProvider, ImmutableMap.of()), read, stream), executor); diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java index ebc753333ae5..cf7df5384127 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java @@ -18,9 +18,12 @@ import static com.google.cloud.storage.ByteSizeConstants._2MiB; import static com.google.cloud.storage.ITObjectReadSessionFakeTest.getReadRange; +import static com.google.cloud.storage.ITObjectReadSessionFakeTest.read; import static com.google.cloud.storage.PackagePrivateMethodWorkarounds.maybeGetStorageDataClient; +import static com.google.cloud.storage.TestUtils.xxd; import static com.google.common.truth.Truth.assertThat; import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertThrows; import com.google.api.gax.grpc.GrpcCallContext; import com.google.cloud.storage.ITObjectReadSessionFakeTest.FakeStorage; @@ -34,10 +37,14 @@ import com.google.storage.v2.Object; import com.google.storage.v2.ObjectRangeData; import io.grpc.stub.StreamObserver; +import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; import org.junit.Test; @@ -66,6 +73,8 @@ public final class ITFastOpenFakeTest { private static final ReadAsFutureBytes READ_10_TO_20 = ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10, 10)); + private static final ReadAsFutureBytes READ_20_TO_30 = + ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(20, 10)); /** Every request the fake server received, in order. */ private final List requestsReceived = @@ -104,6 +113,161 @@ public void openRequest_withoutFastOpenReads_isUnchanged() throws Exception { assertThat(requestsReceived).containsExactly(OPEN); } + @Test + public void readAs_withTheFastOpenConfig_doesNotSendAnotherRequest() throws Exception { + FakeStorage server = + fakeServer(ImmutableMap.of(OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10))); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + try (ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of(READ_10_TO_20)).get(5, SECONDS)) { + byte[] actual = session.readAs(READ_10_TO_20).get(2, SECONDS); + assertThat(xxd(actual)).isEqualTo(xxd(slice(10, 10))); + } + } + + assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20); + } + + @Test + public void readAs_withTheFastOpenConfigAgain_sendsANewRead() throws Exception { + BidiReadObjectRequest secondReadOf10To20 = read(2, 10, 10); + FakeStorage server = + fakeServer( + ImmutableMap.of( + OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10), + secondReadOf10To20, bytes(2, 10, 10))); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + try (ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of(READ_10_TO_20)).get(5, SECONDS)) { + byte[] fastOpen = session.readAs(READ_10_TO_20).get(2, SECONDS); + byte[] second = session.readAs(READ_10_TO_20).get(2, SECONDS); + assertThat(xxd(fastOpen)).isEqualTo(xxd(slice(10, 10))); + assertThat(xxd(second)).isEqualTo(xxd(slice(10, 10))); + } + } + + assertThat(requestsReceived) + .containsExactly(OPEN_WITH_READ_OF_10_TO_20, secondReadOf10To20) + .inOrder(); + } + + @Test + public void readAs_withADifferentConfig_sendsANewRead_andKeepsTheFastOpenRead() throws Exception { + BidiReadObjectRequest readOf20To30 = read(2, 20, 10); + FakeStorage server = + fakeServer( + ImmutableMap.of( + OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10), + readOf20To30, bytes(2, 20, 10))); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + try (ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of(READ_10_TO_20)).get(5, SECONDS)) { + byte[] other = session.readAs(READ_20_TO_30).get(2, SECONDS); + byte[] fastOpen = session.readAs(READ_10_TO_20).get(2, SECONDS); + assertThat(xxd(other)).isEqualTo(xxd(slice(20, 10))); + assertThat(xxd(fastOpen)).isEqualTo(xxd(slice(10, 10))); + } + } + + assertThat(requestsReceived) + .containsExactly(OPEN_WITH_READ_OF_10_TO_20, readOf20To30) + .inOrder(); + } + + @Test + public void close_withAnUnclaimedFastOpenRead_closesTheSession() throws Exception { + FakeStorage server = + fakeServer(ImmutableMap.of(OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10))); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of(READ_10_TO_20)).get(5, SECONDS); + session.close(); + + assertThrows(IllegalStateException.class, () -> session.readAs(READ_10_TO_20)); + } + + assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20); + } + + @Test + public void close_withAnUnclaimedChannelFastOpenRead_doesNotHang() throws Exception { + BidiReadObjectRequest openWithWholeObjectRead = + OPEN.toBuilder().addReadRanges(getReadRange(1, 0, 0)).build(); + FakeStorage server = + FakeStorage.of( + ImmutableMap.of( + openWithWholeObjectRead, + client -> { + client.onNext(metadataOnly()); + // more chunks than a channel read keeps while nobody reads it + for (int offset = 0; offset < 64; offset += 8) { + client.onNext( + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData( + ChecksummedTestContent.of(ALL_OBJECT_BYTES, offset, 8) + .asChecksummedData()) + .setReadRange(getReadRange(1, offset, 8))) + .build()); + } + })); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + ObjectReadSession session = + dataClient + .readSession(OPEN, CTX, ImmutableList.of(ReadProjectionConfigs.asChannel())) + .get(5, SECONDS); + // let the chunks arrive and fill the unread channel read + Thread.sleep(500); + + CompletableFuture close = + CompletableFuture.runAsync( + () -> { + try { + session.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }); + close.get(10, SECONDS); + } + } + + @Test + public void readSession_withMoreThanOneFastOpenConfig_isRejected() throws Exception { + FakeStorage server = fakeServer(ImmutableMap.of()); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + assertThrows( + IllegalArgumentException.class, + () -> dataClient.readSession(OPEN, CTX, ImmutableList.of(READ_10_TO_20, READ_20_TO_30))); + } + + assertThat(requestsReceived).isEmpty(); + } + /** * A fake server that answers each request in {@code responses} and adds it to {@link * #requestsReceived}. Any other request is answered with an error. @@ -138,4 +302,13 @@ private static BidiReadObjectResponse metadataAndBytes(int readId, int offset, i .setRangeEnd(true)) .build(); } + + /** The response to a read request of {@code [offset, offset + length)}. */ + private static BidiReadObjectResponse bytes(int readId, int offset, int length) { + return metadataAndBytes(readId, offset, length).toBuilder().clearMetadata().build(); + } + + private static byte[] slice(int offset, int length) { + return Arrays.copyOfRange(ALL_OBJECT_BYTES, offset, offset + length); + } }