feat(gax): implement baseline Callable and Future for resumable uploads - #14241
feat(gax): implement baseline Callable and Future for resumable uploads#14241whowes wants to merge 1 commit into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
554caca to
b8fc38f
Compare
|
/gemini review |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
234e79f to
6845669
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces the concrete implementation of ResumableUploadCallable and ResumableUploadFuture (ResumableUploadCallableImpl and ResumableUploadFutureImpl) to coordinate resumable upload sessions and stream chunks asynchronously, along with comprehensive unit tests. The feedback highlights a potential issue where performing blocking I/O (ByteStreams.read) inside asynchronous future callbacks could lead to thread starvation or deadlocks if a limited executor is used, suggesting either documenting executor requirements or offloading the blocking read.
| byte[] buffer = new byte[chunkSize]; | ||
| int bytesRead; | ||
| try { | ||
| bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize); |
There was a problem hiding this comment.
Performing blocking I/O (ByteStreams.read) inside asynchronous future callbacks (which run on the provided executor) can lead to thread starvation or deadlocks if the executor is a direct executor or a limited thread pool (such as gRPC network threads). Consider documenting that the executor passed to the callable must be a dedicated thread pool suitable for blocking I/O operations, or offloading the blocking read to a dedicated I/O executor.
6845669 to
61fd370
Compare
| ResumableUploadCallSettings effectiveSettings = defaultCallSettings.merge(settings); | ||
|
|
||
| return ResumableUploadFutureImpl.create( | ||
| client, request, payload, effectiveSettings.getChunkSize(), defaultCallContext, executor); |
There was a problem hiding this comment.
Since we know there will be more configurations, can we pass the whole settings class to the future?
| + " incomplete status")); | ||
| } | ||
| // Continuation: asynchronously transmit subsequent chunk with updated offset. | ||
| return transmitChunks( |
There was a problem hiding this comment.
Can we use a while loop instead of recursive calls? There is always stackoverflow concerns using recursives.
| return transmitChunks(client, payload, chunkSize, callContext, url, 0L, executor); | ||
| }, | ||
| executor); | ||
| sessionFuture.addListener(() -> closePayload(payload), executor); |
There was a problem hiding this comment.
I think there are two issues here:
- Should we take the responsibility of closing the stream? Usually whoever creates the stream is responsible for it.
- If we do want to take the responsibility, using try-with-resources is preferred than manually closing it.
| return transmitChunks( | ||
| client, payload, chunkSize, callContext, uploadSessionUrl, nextOffset, executor); | ||
| }, | ||
| executor); |
There was a problem hiding this comment.
I think the whole upload(including the initial call) can be done in a single thread. Using transformAsync may transform the future in a different thread, we can ended up with a lot of thread when uploading large files.
| private static final byte[] EMPTY_PAYLOAD = new byte[0]; | ||
|
|
||
| private final InputStream payload; | ||
| private final AtomicReference<@Nullable String> uploadSessionUrl; |
There was a problem hiding this comment.
ResumableUploadFuture represents one main upload session and there should be only one thread modifying this url. I don't think we need to use AtomicReference.
There was a problem hiding this comment.
In general, I think the current structure of the how we make initial call and upload call can be improved. The nested calls of ApiFutures.transformAsync is not easy to read, and may have performance concerns. Some future calls can be made in the callable as well. There could also be a wrapper callable/future that does the whole uploading.
A pseudo code I'm thinking in the Callable is
StartUploadFuture startUploadFuture = client.startUploadCallable().futureCall();
UploadWholeCallable uploadWholeCallable = new UploadWholeCallable(startUploadFuture, client);
UploadWholeFuture uploadWholeFuture = uploadWholeCallable().futureCall();
return new ResumableUploadFuture(startUploadFuture, uploadWholeFuture).
This is similar to OperationCallableImpl.
Let me know what you think and if I missed anything.
61fd370 to
cf3098c
Compare
cf3098c to
e237724
Compare
|
|
2 participants
This implementation supports the happy path only; retries, recovery, timeouts, per-call settings, and progress tracking will be added in subsequent phases.