blocking stub v2 server streaming calls sometimes require .read() to initiate request

[benjaminp]

If the channel is in an idle state when ClientCalls.blockingV2ServerStreamingCall is invoked, sending the request will be queued up in DelayedClientTransport. When the connection is set up, DelayedClientTransport will run all the deferred request initiation steps on the call executor:

grpc-java/core/src/main/java/io/grpc/internal/DelayedClientTransport.java

Lines 306 to 312 in b88536a

	Executor executor = defaultAppExecutor;
	// createRealStream may be expensive. It will start real streams on the transport. If
	// there are pending requests, they will be serialized too, which may be expensive. Since
	// we are now on transport thread, we need to offload the work to an executor.
	if (callOptions.getExecutor() != null) {
	executor = callOptions.getExecutor();
	}

Unfortunately, the call executor for the blocking v2 stubs is a ThreadSafeThreadlessExecutor, which means the headers and request aren't actually sent until the client calls .read() on the BlockingClientCall.

In contrast, the blocking v1 stub always sent the request as soon as possible.

[kannanjgithub]

Yes, it uses the calling thread so if you don't call read it might not do the things you expect. What is the pattern you are using here for the RPC? For us to figure out a solution better we would like to know what your interactions with the BlockingClientCall roughly look like. What does your code do after starting the RPC, does it go on to do some other work before it comes back to do the read on the RPC? What is the reason for not immediately calling read, is it because the RPC is now slower ? Is it because you don't care about the result of the read ?

[benjaminp]

The first place I ran into this issue was a test. The test wanted to start an RPC, have the server generate the first response, and then wait. However, with the blocking v2 stub, the request was never sent to the server.

The second place I ran into this was a gathering broadcast pattern roughly like this:

List<Response> gatherResponses(List<ManagedChannel> channels, Request myRequest) {
  var calls = new ArrayList<BlockingClientCall<Request, Response>>
  for (var ch : channels) {
    calls.add(MyStub.newBlockingV2Stub().myServerStreamingRpc(myRequest));
  }
  var responses = new ArrayList<Response>();
  for (var call : calls) {
     responses.add(call.read());
  }
  return responses;
}

If one of the later channels in the channels list wasn't already initialized, it takes until many other requests were completed before finally start that channel's request.

I find the largest problem to be that the behavior is inconsistent. It could be reasonable that is.read() is always required to kick off the request.

[ejona86]

There's multiple places this general "the rpc could have made progress but didn't" could be a problem:

1. Channel exiting idle. The RPC waits for service config, meaning waiting for DNS to resolve the address
2. When LB is CONNECTING
3. CallCredentials when doing async work
4. Retries/hedging

Almost all cases this comes up should be at the beginning of the call. Retries/hedging should be the main outlier and can happen at any time before receiving response headers.

For client-streaming and bidi streaming, waiting for onReady() callback to handle most of the cases. Server-streaming has an optimized send path since there's only one message, which only leaves onHeaders() which is pretty late. Both cases have the problem that we don't currently block when starting the RPC.

[kannanjgithub]

I hope your issue is resolved. Feel free to comment to reopen in the future if needed.

[ejona86]

@kannanjgithub, this is a concerning report. We can't guarantee to address it, but it does warrant continued attention.

I talked with dfawley yesterday and grpc-go does not return the stream to the application until a stream is allocated, which means passing through 1-3 on my list and MAX_CONCURRENT_STREAMS as well. Retries require interactions (reads/writes) from the application, and they are okay with that, as there's few alternatives.

I think this problem is unique to server streaming, as ordinarily sending the request waits for the RPC to become ready. Unary and server-streaming optimizations where readiness is basically unavailable, as the RPC itself is delayed until the request and half close is enqueued, to send it all-at-once in one packet. I don't know if isReady() ever returns true for them, but at least onReady() is squelched.

So blocking until the RPC is allocated would need changes gRPC's internals. That could maybe be addressed by only calling onReady() if isReady() has been called and returned false (there's some subtleties around this as applications are certainly depending on onReady() always being called for other streaming types, at least for client-side). But we'd also need to change the method signature to allow interruption, or at least cancel the RPC when interrupted. It may be surprising for creating the stream to block without having InterruptedException; unary has that behavior, but it isn't surprising in that case. Or we could force the application to send the initial request...

[ejona86]

We could share the ThreadSafeThreadlessExecutor across RPCs. I figure we'd let someone create the executor (probably renamed or behind a factor method) and then pass it in via stub.withExecutor(). ClientCalls would detect the type and use that executor instead of creating a new one. But that requires all RPCs that share an executor are only used on a single thread (although still one for reading, one for writing) and you can't change it after they are created.

There's some potential for getting fancy, but I expect that virtual threads will end up providing a better option pretty quickly in the fanciness curve.