Grpc-java: okhttp: isReady does not become false

[pchmielowski]

In Android application I'm trying to upload a large file (165 MB) using client gRPC streaming. The file is chunked into 2 MB chunks.

val channel = AndroidChannelBuilder
    .forAddress("server address", 443)
    .context(this)
    .intercept(GrpcAuthorizationInterceptor())
    .build()

val source = FileInputStream(File("/sdcard/Movies/Movie.mp4"))

FileServiceGrpc.newStub(channel)
    .uploadFile(object : ClientResponseObserver<UploadFileRequest, UploadFileResponse> {
        override fun beforeStart(requestStream: ClientCallStreamObserver<UploadFileRequest>) {
            var counter = 0
            requestStream.setOnReadyHandler {
                while (requestStream.isReady) {
                    Log.d("MainActivity", "sending ${counter++} chunk")
                    val bytes = ByteArray(2 * 1024 * 1024)
                    source.read(bytes)
                    val byteString = ByteString.copyFrom(bytes)
                    val request = UploadFileRequest.newBuilder()
                        .setData(byteString)
                        .setExtension(UploadFileRequest.Extension.MP4)
                        .build()
                    requestStream.onNext(request)
                }
            }
        }

        override fun onNext(value: UploadFileResponse) {
        }

        override fun onError(t: Throwable) = throw t

        override fun onCompleted() {
        }
})

Each time I run this code, the application crashes with:
java.lang.OutOfMemoryError: Failed to allocate a 24 byte allocation with 125656 free bytes and 122KB until OOM, target footprint 201326592, growth limit 201326592; failed due to fragmentation (largest possible contiguous allocation 0 bytes)

Log output before the crash is as following:

sending 0 chunk
...
sending 61 chunk

What is important, there is no delay between each sending x chunk log entry, which means the callback provided to setOnReadyHandler is called immediately and requestStream.isReady flag stays true.

Seems like here is the problem: even though the previous data has not been sent and is buffered by gRPC, isReady returns true. The JavaDoc of isReady method says the following:

   * If {@code true}, indicates that the observer is capable of sending additional messages
   * without requiring excessive buffering internally. This value is just a suggestion and the
   * application is free to ignore it, however doing so may result in excessive buffering within the
   * observer.

Therefore my understanding is, that in this case false should be returned, because OutOfMemoryError suggests, that excessive buffering is happening.

How can I improve the code to get rid of OutOfMemoryError? Also is my understanding of isReady() method correct?

[voidzcy]

You need to disable auto flow control, a.k.a., auto requests (see how the manual flow control example works). Without it, you do not get the back pressure.

[pchmielowski]

The code behaves in exactly the same way after adding requestStream.disableAutoRequestWithInitial(1) line.

Isn't it intended to use only of a streaming response? The comment says: Set up manual flow control for the response stream.
In my situation there is streaming request and a single response.
Here is a definition of the service:

rpc UploadFile (stream UploadFileRequest) returns (UploadFileResponse);

[ejona86]

@pchmielowski, can you make sure onError() isn't being called? I think the isReady "fails open" if the RPC is killed (which was done to avoid deadlocks, but is probably the wrong behavior).

[pchmielowski]

@ejona86 No. I've double checked it, but onError is not called.

[ejona86]

@voidzcy, could you try to reproduce? Maybe by hacking up the manual flow control example?

disableAutoRequestWithInitial is for the inbound direction, not outbound. isReady/onReady is the thing to look at.

[voidzcy]

It might be hard to reproduce the OutOfMemoryError, I will see the behavior of isReady bit. As the @pchmielowski observes that isReady stays true while messages are not sent out.

[ejona86]

We don't need to reproduce the OOME. If isReady isn't changing, then we have something to fix.

[voidzcy]

It turns out isReady works fine with okhttp. I do not reproduce the observation that isReady bit stays true. Here is how I experimented it:

Based on manual flow control example, I modified the service definition to

service ServerStreamingGreeter {
  rpc SayHello(stream HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string name = 1;
  bytes data = 2;
}

message HelloReply {
  string message = 1;
}

The only thing changed on the server side is not to send responses on in request observer's onNext(). Client side is basically the same except adding 2MB bytes to each request and use an okhttp channel. For most of the time, isReady bit is false. From the log I can see request messages are sent out every 2 seconds.

@pchmielowski Did you do any thing else in your code? Also which grpc version are you using (I used the master head in my experiment)?

[pchmielowski]

Here is a list of dependencies:

    def grpc_version = "1.30.0"
    implementation "io.grpc:grpc-okhttp:$grpc_version"
    implementation "io.grpc:grpc-android:$grpc_version"
    implementation "io.grpc:grpc-protobuf-lite:$grpc_version"
    implementation "io.grpc:grpc-stub:$grpc_version"

I also have the following interceptor:

internal class GrpcAuthorizationInterceptor : ClientInterceptor {

    override fun <ReqT, RespT> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ) = object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
            next.newCall(method, callOptions)
        ) {
            override fun start(responseListener: Listener<RespT>, headers: Metadata) {
                headers.addToken()
                super.start(responseListener, headers)
            }
        }

    private fun Metadata.addToken() {
        val token = "abcd"
        put(Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER), "Bearer $token")
    }
}

[pchmielowski]

I'm observing the same behaviour after changing to OkHttpChannelBuilder.

Also removing:

val source = FileInputStream(File("/sdcard/Movies/Movie.mp4"))

from my original post and adding:

val source = object : InputStream() {
    override fun read() = 0
}

causes the same behaviour.

Here are logs with the time signatures so you can see there is almost no delay between sending each chunks:

2020-06-25 12:56:10.403 D: sending 0 chunk
2020-06-25 12:56:10.469 D: sending 1 chunk
2020-06-25 12:56:10.486 D: sending 2 chunk
2020-06-25 12:56:10.502 D: sending 3 chunk
2020-06-25 12:56:10.518 D: sending 4 chunk
2020-06-25 12:56:10.534 D: sending 5 chunk
2020-06-25 12:56:10.550 D: sending 6 chunk
2020-06-25 12:56:10.566 D: sending 7 chunk
2020-06-25 12:56:10.582 D: sending 8 chunk
2020-06-25 12:56:10.598 D: sending 9 chunk
2020-06-25 12:56:10.612 D: sending 10 chunk
2020-06-25 12:56:10.628 D: sending 11 chunk
2020-06-25 12:56:10.642 D: sending 12 chunk
2020-06-25 12:56:10.657 D: sending 13 chunk
2020-06-25 12:56:10.674 D: sending 14 chunk
2020-06-25 12:56:10.690 D: sending 15 chunk
2020-06-25 12:56:10.712 D: sending 16 chunk
2020-06-25 12:56:10.732 D: sending 17 chunk
2020-06-25 12:56:10.748 D: sending 18 chunk
2020-06-25 12:56:10.762 D: sending 19 chunk
2020-06-25 12:56:10.776 D: sending 20 chunk

[voidzcy]

This is extremely weird. I used almost exactly the same code as you did, but the log shows messages are sent slowly:

Jun 25, 2020 4:57:59 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 0 chunk
Jun 25, 2020 4:57:59 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 1 chunk
Jun 25, 2020 4:57:59 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 2 chunk
Jun 25, 2020 4:57:59 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 3 chunk
Jun 25, 2020 4:58:00 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 4 chunk
Jun 25, 2020 4:58:01 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 5 chunk
Jun 25, 2020 4:58:02 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 6 chunk
Jun 25, 2020 4:58:03 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 7 chunk
Jun 25, 2020 4:58:04 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 8 chunk
Jun 25, 2020 4:58:05 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 9 chunk
Jun 25, 2020 4:58:06 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 10 chunk
Jun 25, 2020 4:58:07 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 11 chunk
Jun 25, 2020 4:58:08 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 12 chunk
Jun 25, 2020 4:58:09 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 13 chunk
Jun 25, 2020 4:58:10 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 14 chunk
Jun 25, 2020 4:58:11 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 15 chunk
Jun 25, 2020 4:58:12 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 16 chunk
Jun 25, 2020 4:58:13 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 17 chunk
Jun 25, 2020 4:58:14 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 18 chunk
Jun 25, 2020 4:58:15 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 19 chunk
Jun 25, 2020 4:58:16 PM io.grpc.examples.manualflowcontrol.ManualFlowControlClient$2$1 run
INFO: Sending 20 chunk

Now, the only difference between your code and my experiment is you are using Kotlin and running on Android. But just based on the code snippets you provided, I don't see anything that could go wrong with Kotlin and Android. Are you able to provide a minimal reproducible project so that we can run it and see what on earth is happening?

[pchmielowski]

I'm connecting to proprietary server so I can not provide a project with the credentials.
However I'll try to adapt example server from the examples directory to reproduce this situation and provide you both the client and server code in a few days.

[pchmielowski]

Unfortunately I haven't found a time to properly reproduce the issue - for now we changed the API contract and instead of streaming, we are using separate unary calls to send each chunk of data.
Therefore for now I'm closing the issue - probably later this year we will try with streaming once again, so if the issue still occur and I'm able to create full code to reproduce it, I'll reopen it (or open a new) with the code.