feat(gax): implement baseline Callable and Future for resumable uploads - #14241

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.

Since we know there will be more configurations, can we pass the whole settings class to the future?

Done.

Can we use a while loop instead of recursive calls? There is always stackoverflow concerns using recursives.

IIUC a conventional while would pretty much map to the single thread per upload idea (and keeping the thread pinned to the upload even while it's blocking on I/O)? That seems problematic to me (see other comment).

On the recursion point (this particular code is restructured, but the new code chains kinda similarly): it looks recursion-ish, but stack frames don't actually accumulate. futureCall returns immediately, transmitChunk returns, and the transmitChunk stack frame is popped. When the chunk future finishes later, the executor invokes the callback and it's not coupled with the stack frame of when it was scheduled. (The overflow miiight be a risk if the executor used here was a DirectExecutor which was possible in the last snapshot, but I switched to a ScheduledExecutorService as I think we'll need that when we layer in retries.)

IIUC this callback chaining pattern is pretty similar to CallbackChainRetryingFuture which does a loop inside its completion listener (submit -> setAttemptFuture -> attach listener -> repeat) across retry attempts rather than blocking a thread in a loop.

SGTM. I agree that there are no recursive concerns also.

My original idea is to use an IO thread (which has a large thread pool) for the whole operation since each low-level call is done in a separate thread anyway, this can simplify certain implementation such as updating/reading upload url, reusing the same byte[] etc.. But this requires us to have a "main thread" for each upload session that monitors the progress of whole operation, which still poses risks of thread starvation if we have a lot of uploads.

I think there are two issues here:

1. Should we take the responsibility of closing the stream? Usually whoever creates the stream is responsible for it.
2. If we do want to take the responsibility, using try-with-resources is preferred than manually closing it.

Typically the caller that provides the resource is responsible for closing it in synchronous code, but it's problematic in async calls when using try-with-resources:

try (InputStream stream = getStream()) {
  callable.futureCall(request, stream);
} // <-- stream closed
future.get(); <-- possible failure if callable tried to use stream when closed

This is an issue for both 1 and 2 IMO:

1. with the typical convenient approach not so safe and easy for async it's on the caller to figure out when it's safe to close the stream, which can be tricky to keep track of particularly if the result is accessed far away (in code) from where the stream was created and passed along. So having responsibility transferred to the async task - provided that fact is clearly documented - removes that cognitive load from the caller.

2. On the future impl side if we are making our impl mostly asynchronous rather than writing a synchronous while-style with a dedicated thread per upload (which IMO we should do, the theme of several of my other comments :) then we suffer from the same problem of simple try-with-resources closing the streams prematurely.

Another idea is to change the input from a raw stream to a supplier of stream. So that Gax takes the responsibility of opening and closing the stream completely.
That being said, it does not block this PR and can be refactored later.

That makes sense - possibly this could be additive and the existing version that takes an InputStream directly could then be retrofit to use a supplier that returns the provided instance (requiring all callers to use a supplier may be a bit inconvenient for non-power users). Let's revisit this later.

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.

True, there is only one writer so the AtomicReference is probably overkill (though I don't think it adds much overhead). I switched to a regular @Nullable field but I believe it needs to be volatile since multiple threads may read it (which could happen via getUploadSessionUrl()).

SGTM. I agree that there are no recursive concerns also.

My original idea is to use an IO thread (which has a large thread pool) for the whole operation since each low-level call is done in a separate thread anyway, this can simplify certain implementation such as updating/reading upload url, reusing the same byte[] etc.. But this requires us to have a "main thread" for each upload session that monitors the progress of whole operation, which still poses risks of thread starvation if we have a lot of uploads.

Another idea is to change the input from a raw stream to a supplier of stream. So that Gax takes the responsibility of opening and closing the stream completely.
That being said, it does not block this PR and can be refactored later.

For follow up PRs, I think we need to give it more thought how to close the stream in cancel. Because the stream might be processing in another chunk uploading thread.

Acknowledged.

For follow up PRs, we should think about how to handle this error case. I don't think we should ignore it.

Acknowledged.