Skip to content

Commit 7bd73d3

Browse files
BenWhiteheadBenWhitehead
authored
chore: merge train for nonblocking-appendable/** branches (#3231)
## Description feat: *breaking behavior* rewrite Storage.blobAppendableUpload to be non-blocking and have improved throughput (#3231) Rewrite internals of BlobAppendableUpload to provide non-blocking write calls, and it take advantage of grpc async message handling. When `AppendableUploadWriteableByteChannel#write(ByteBuffer)` is called, an attempt will be made to enqueue the bytes in the outbound queue to GCS. If there is only enough room to partially consume the bytes provided in the `ByteBuffer` the write call will return early specifying the number of bytes actually consumed. As acknowledgements come in from gcs, enqueued messages will be evicted freeing space in the outbound queue. Thereby allowing more bytes to be consumed and enqueued. Given appendable objects are still in private preview I can't quote any metrics here, however preliminary benchmarking of several million objects across a range of sizes show across the board throughput improvments. Because the channel's write call is now non-blocking, if you want to block your application until the full buffer is consumed some new helper methods have been added in StorageChannelUtils to provide blocking behavior. A new method `MinFlushSizeFlushPolicy#withMaxPendingBytes(long)` has been added to allow limiting the number of pending outbound bytes. The default values is 16MiB, but can be configured lower if necessary. ## Release Notes BEGIN_COMMIT_OVERRIDE BEGIN_NESTED_COMMIT feat: *breaking behavior* rewrite Storage.blobAppendableUpload to be non-blocking and have improved throughput (#3231) END_NESTED_COMMIT BEGIN_NESTED_COMMIT feat: add StorageChannelUtils to provide helper methods to perform blocking read/write to/from non-blocking channels (#3231) END_NESTED_COMMIT BEGIN_NESTED_COMMIT feat: add MinFlushSizeFlushPolicy#withMaxPendingBytes(long) (#3231) END_NESTED_COMMIT BEGIN_NESTED_COMMIT fix: update BlobAppendableUploadConfig and FlushPolicy.MinFlushSizeFlushPolicy to default to 4MiB minFlushSize and 16MiB maxPendingBytes (#3249) END_NESTED_COMMIT BEGIN_NESTED_COMMIT fix: make FlushPolicy${Min,Max}FlushSizeFlushPolicy constructors private (#3217) END_NESTED_COMMIT END_COMMIMT_OVERRIDE ## Sub PRs This PR is made of up the following PRs, in sequence 1. #3217 2. #3218 3. #3219 4. #3220 5. #3221 6. #3222 7. #3223 8. #3224 9. #3225 10. #3226 11. #3227 12. #3228 13. #3229 14. #3230 15. #3235 16. #3236 17. #3241 18. #3242 19. #3246 20. #3248 21. #3249 22. #3252
1 parent 5240d03 commit 7bd73d3

File tree

54 files changed

+6906
-2585
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

54 files changed

+6906
-2585
lines changed

‎google-cloud-storage/clirr-ignored-differences.xml‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,4 +161,27 @@
161161
<method>com.google.cloud.storage.BucketInfo$Builder setGoogleManagedEncryptionEnforcementConfig(com.google.cloud.storage.BucketInfo$GoogleManagedEncryptionEnforcementConfig)</method>
162162
</difference>
163163

164+
<!-- make beta api constructors private, they still retain their factory methods. -->
165+
<difference>
166+
<differenceType>7004</differenceType>
167+
<className>com/google/cloud/storage/FlushPolicy$MinFlushSizeFlushPolicy</className>
168+
<method>FlushPolicy$MinFlushSizeFlushPolicy(int)</method>
169+
</difference>
170+
<difference>
171+
<differenceType>7009</differenceType>
172+
<className>com/google/cloud/storage/FlushPolicy$MinFlushSizeFlushPolicy</className>
173+
<method>FlushPolicy$MinFlushSizeFlushPolicy(int)</method>
174+
</difference>
175+
<difference>
176+
<differenceType>7009</differenceType>
177+
<className>com/google/cloud/storage/FlushPolicy$MaxFlushSizeFlushPolicy</className>
178+
<method>FlushPolicy$MaxFlushSizeFlushPolicy(int)</method>
179+
</difference>
180+
<!-- this method is already on the parent interface, and is on an @InternalExtensionOnly interface -->
181+
<difference>
182+
<differenceType>7012</differenceType>
183+
<className>com/google/cloud/storage/BlobAppendableUpload$AppendableUploadWriteableByteChannel</className>
184+
<method>int write(java.nio.ByteBuffer)</method>
185+
</difference>
186+
164187
</differences>

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

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,195 @@
1+
/*
2+
* Copyright 2025 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.storage;
18+
19+
import com.google.cloud.BaseServiceException;
20+
import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
21+
import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
22+
import java.io.IOException;
23+
import java.io.InterruptedIOException;
24+
import java.nio.ByteBuffer;
25+
import java.nio.channels.ClosedChannelException;
26+
import java.util.concurrent.ExecutionException;
27+
import java.util.concurrent.TimeUnit;
28+
import java.util.concurrent.TimeoutException;
29+
30+
final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
31+
32+
private final BidiUploadStreamingStream stream;
33+
private final ChunkSegmenter chunkSegmenter;
34+
35+
private boolean open;
36+
private long writeOffset;
37+
private volatile boolean nextWriteShouldFinalize;
38+
private boolean writeCalledAtLeastOnce;
39+
40+
/** If write throws an error, don't attempt to finalize things when {@link #close()} is called. */
41+
private boolean writeThrewError;
42+
43+
BidiAppendableUnbufferedWritableByteChannel(
44+
BidiUploadStreamingStream stream, ChunkSegmenter chunkSegmenter, long writeOffset) {
45+
this.stream = stream;
46+
this.chunkSegmenter = chunkSegmenter;
47+
this.open = true;
48+
this.writeOffset = writeOffset;
49+
this.nextWriteShouldFinalize = false;
50+
this.writeThrewError = false;
51+
}
52+
53+
@Override
54+
public long write(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
55+
return internalWrite(srcs, srcsOffset, srcsLength);
56+
}
57+
58+
@Override
59+
public long writeAndClose(ByteBuffer[] srcs, int offset, int length) throws IOException {
60+
long totalRemaining = Buffers.totalRemaining(srcs, offset, length);
61+
// internalWrite is non-blocking, but close is blocking.
62+
// loop here to ensure all the bytes we need flush are enqueued before we transition to trying
63+
// to close.
64+
long written = 0;
65+
do {
66+
written += internalWrite(srcs, offset, length);
67+
} while (written < totalRemaining);
68+
close();
69+
return written;
70+
}
71+
72+
@Override
73+
public boolean isOpen() {
74+
return open;
75+
}
76+
77+
@Override
78+
public void close() throws IOException {
79+
if (!open) {
80+
return;
81+
}
82+
try {
83+
if (writeThrewError) {
84+
return;
85+
}
86+
87+
if (!writeCalledAtLeastOnce) {
88+
stream.flush();
89+
}
90+
if (nextWriteShouldFinalize) {
91+
//noinspection StatementWithEmptyBody
92+
while (!stream.finishWrite(writeOffset)) {}
93+
} else {
94+
//noinspection StatementWithEmptyBody
95+
while (!stream.closeStream(writeOffset)) {}
96+
}
97+
98+
awaitResultFuture();
99+
} finally {
100+
stream.sendClose();
101+
open = false;
102+
}
103+
}
104+
105+
public void nextWriteShouldFinalize() {
106+
this.nextWriteShouldFinalize = true;
107+
}
108+
109+
private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
110+
if (!open) {
111+
throw new ClosedChannelException();
112+
}
113+
// error early. if the result future is already failed, await it to throw the error
114+
if (stream.getResultFuture().isDone()) {
115+
awaitResultFuture();
116+
return 0;
117+
}
118+
writeCalledAtLeastOnce = true;
119+
120+
long availableCapacity = stream.availableCapacity();
121+
if (availableCapacity <= 0) {
122+
return 0;
123+
}
124+
RewindableContent rewindableContent = RewindableContent.of(srcs, srcsOffset, srcsLength);
125+
long totalBufferRemaining = rewindableContent.getLength();
126+
127+
ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength, true);
128+
if (data.length == 0) {
129+
return 0;
130+
}
131+
// we consumed some bytes from srcs, flag our content as dirty since we aren't writing
132+
// those bytes to implicitly flag as dirty.
133+
rewindableContent.flagDirty();
134+
135+
long bytesConsumed = 0;
136+
for (int i = 0, len = data.length, lastIdx = len - 1; i < len; i++) {
137+
ChunkSegment datum = data[i];
138+
int size = datum.getB().size();
139+
boolean appended;
140+
if (i < lastIdx) {
141+
appended = stream.append(datum);
142+
} else if (i == lastIdx && nextWriteShouldFinalize) {
143+
appended = stream.appendAndFinalize(datum);
144+
} else {
145+
appended = stream.appendAndFlush(datum);
146+
}
147+
if (appended) {
148+
bytesConsumed += size;
149+
writeOffset += size;
150+
} else {
151+
// if we weren't able to trigger a flush by reaching the end of the array and calling
152+
// appendAndFlush, explicitly call flush here so that some progress can be made.
153+
// we prefer appendAndFlush so a separate message is not needed, but an extra message
154+
// in order to make progress and free buffer space is better than ending up in a live-lock.
155+
stream.flush();
156+
break;
157+
}
158+
}
159+
160+
if (bytesConsumed != totalBufferRemaining) {
161+
rewindableContent.rewindTo(bytesConsumed);
162+
}
163+
164+
return bytesConsumed;
165+
}
166+
167+
private void awaitResultFuture() throws IOException {
168+
try {
169+
stream.getResultFuture().get(10_717, TimeUnit.MILLISECONDS);
170+
} catch (InterruptedException e) {
171+
Thread.currentThread().interrupt();
172+
InterruptedIOException ioe = new InterruptedIOException();
173+
ioe.initCause(e);
174+
writeThrewError = true;
175+
throw ioe;
176+
} catch (ExecutionException e) {
177+
BaseServiceException coalesce = StorageException.coalesce(e.getCause());
178+
String message = coalesce.getMessage();
179+
String ioExceptionMessage = message;
180+
// if the failure is an upload scenario we detect client side, it's message will be
181+
// verbose. To avoid duplication, select the first line only for the io exception
182+
int firstNewLineIndex = message != null ? message.indexOf('\n') : -1;
183+
if (firstNewLineIndex > -1) {
184+
ioExceptionMessage = message.substring(0, firstNewLineIndex);
185+
}
186+
IOException ioException = new IOException(ioExceptionMessage, coalesce);
187+
// ioException.addSuppressed(new AsyncStorageTaskException());
188+
writeThrewError = true;
189+
throw ioException;
190+
} catch (TimeoutException e) {
191+
writeThrewError = true;
192+
throw new IOException(e);
193+
}
194+
}
195+
}

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,8 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(
111111
GrpcStorageImpl grpc = (GrpcStorageImpl) s;
112112
GrpcCallContext grpcCallContext =
113113
opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
114-
BidiWriteObjectRequest req = grpc.getBidiWriteObjectRequest(info, opts);
114+
BidiWriteObjectRequest req =
115+
grpc.getBidiWriteObjectRequest(info, opts, false);
115116

116117
ApiFuture<BidiResumableWrite> startResumableWrite =
117118
grpc.startResumableWrite(grpcCallContext, req, opts);

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

Lines changed: 0 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -94,73 +94,3 @@ static BidiResumableWrite identity(BidiResumableWrite w) {
9494
return w;
9595
}
9696
}
97-
98-
final class BidiAppendableWrite implements BidiWriteObjectRequestBuilderFactory {
99-
100-
private final BidiWriteObjectRequest req;
101-
102-
public BidiAppendableWrite(BidiWriteObjectRequest req) {
103-
this(req, false);
104-
}
105-
106-
public BidiAppendableWrite(BidiWriteObjectRequest req, boolean takeOver) {
107-
if (takeOver) {
108-
this.req = req;
109-
} else {
110-
req =
111-
req.toBuilder()
112-
.setWriteObjectSpec(req.getWriteObjectSpec().toBuilder().setAppendable(true).build())
113-
.build();
114-
this.req = req;
115-
}
116-
}
117-
118-
public BidiWriteObjectRequest getReq() {
119-
return req;
120-
}
121-
122-
@Override
123-
public BidiWriteObjectRequest.Builder newBuilder() {
124-
return req.toBuilder();
125-
}
126-
127-
@Override
128-
public @Nullable String bucketName() {
129-
if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
130-
return req.getWriteObjectSpec().getResource().getBucket();
131-
} else if (req.hasAppendObjectSpec()) {
132-
return req.getAppendObjectSpec().getBucket();
133-
}
134-
return null;
135-
}
136-
137-
@Override
138-
public String toString() {
139-
return "BidiAppendableWrite{" + "req=" + fmtProto(req) + '}';
140-
}
141-
142-
@Override
143-
public boolean equals(Object o) {
144-
if (this == o) {
145-
return true;
146-
}
147-
if (!(o instanceof BidiAppendableWrite)) {
148-
return false;
149-
}
150-
BidiAppendableWrite BidiAppendableWrite = (BidiAppendableWrite) o;
151-
return Objects.equals(req, BidiAppendableWrite.getReq());
152-
}
153-
154-
@Override
155-
public int hashCode() {
156-
return Objects.hash(req);
157-
}
158-
159-
/**
160-
* Helper function which is more specific than {@link Function#identity()}. Constraining the input
161-
* and output to be exactly {@link BidiAppendableWrite}.
162-
*/
163-
static BidiAppendableWrite identity(BidiAppendableWrite w) {
164-
return w;
165-
}
166-
}

0 commit comments

Comments
 (0)