Repository navigation
feat(common): add StartWithCorkedInitialMetadata. - #16569
Conversation
`StartWithCorkedInitialMetadata()` allows callers that immediately send an initial `Write()` after starting a bidirectional stream to save a network send and a `grpc::CompletionQueue::Next()` call on stream startup. When a stream is started after calling `grpc::ClientContext::set_initial_metadata_corked(true)`, it does not send the initial metadata until the first `Write()` call, coalescing it with the first message's network write. It also does not bounce a tag for starting the stream through the CompletionQueue. RPCs that can tolerate this behavior have fewer thread hops and less contention on `grpc::CompletionQueue::Next()`. Unfortunately there's no getter for initial metadata in the `grpc::ClientContext` (and anyway we don't plumb the `ClientContext` into `google::cloud::internal::CompletionQueueImpl`). So we have to implement this at a higher level where the `ClientContext` is available. I personally will use this to optimize bidirectional streaming RPCs, so I put it in `AsyncStreamingReadWriteRpc` for the purposes of review. But analogous and mechanical changes could be done in `AsyncStreamingReadRpc` and `AsyncStreamingWriteRpc`.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the StartWithCorkedInitialMetadata() method to the AsyncStreamingReadWriteRpc interface and implements it across various stream decorators, wrappers, and mock classes. This allows starting a streaming RPC with corked initial metadata, delaying its transmission until the first write call. The review feedback suggests simplifying the template arguments in the tracing wrapper by using the base class name directly instead of decltype(impl_)::element_type, and explicitly typing the lambda parameter in the timeout wrapper instead of using auto to improve readability.
| future<bool> Start() override { | ||
| // It is sufficient to set `span_` as the parent of `start_span`, because | ||
| // the lower levels do not create any spans. | ||
| opentelemetry::trace::StartSpanOptions options; | ||
| options.parent = span_->GetContext(); | ||
| auto start_span = internal::MakeSpan("Start", options); | ||
| return impl_->Start().then( | ||
| [this, ss = std::move(start_span)](future<bool> f) { | ||
| EndSpan(*ss); | ||
| auto started = f.get(); | ||
| span_->SetAttribute("gl-cpp.stream_started", started); | ||
| started_ = started; | ||
| return started; | ||
| }); | ||
| return StartInternal<&decltype(impl_)::element_type::Start>("Start"); | ||
| } | ||
|
|
||
| future<bool> StartWithCorkedInitialMetadata() override { | ||
| return StartInternal< | ||
| &decltype(impl_)::element_type::StartWithCorkedInitialMetadata>( | ||
| "StartWithCorkedInitialMetadata"); | ||
| } |
There was a problem hiding this comment.
Using decltype(impl_)::element_type is unnecessarily verbose and complex. Since AsyncStreamingReadWriteRpcTracing inherits from AsyncStreamingReadWriteRpc<Request, Response>, you can refer to the base class type directly, which is much cleaner and more readable.
future<bool> Start() override {
return StartInternal<&AsyncStreamingReadWriteRpc<Request, Response>::Start>("Start");
}
future<bool> StartWithCorkedInitialMetadata() override {
return StartInternal<
&AsyncStreamingReadWriteRpc<Request, Response>::StartWithCorkedInitialMetadata>(
"StartWithCorkedInitialMetadata");
}References
- Reject "Code Poetry": Dismantle complex abstractions used for simple tasks. Prefer simplicity over cleverness. (link)
There was a problem hiding this comment.
I disagree that it is "less verbose" to spell decltype(impl_)::element_type as AsyncStreamingReadWriteRpc<Request, Response> (30 chars -> 46 chars), but I agree that decltype is unnecessary and can confuse people. I've reworded to the base class name.
| future<bool> StartWithCorkedInitialMetadata() { | ||
| auto watchdog = CreateWatchdog(start_timeout); | ||
| return child->StartWithCorkedInitialMetadata().then( | ||
| [watchdog = std::move(watchdog), w = WeakFromThis()](auto f) mutable { | ||
| if (auto self = w.lock()) { | ||
| return self->OnStart(std::move(watchdog), f.get()); | ||
| } | ||
| return make_ready_future(false); | ||
| }); | ||
| } |
There was a problem hiding this comment.
Avoid using auto for the lambda parameter f here. Specifying the explicit type future<bool> is more readable, avoids creating a generic lambda unnecessarily, and is consistent with the other stream wrappers in this PR (such as the tracing and logging wrappers).
future<bool> StartWithCorkedInitialMetadata() {
auto watchdog = CreateWatchdog(start_timeout);
return child->StartWithCorkedInitialMetadata().then(
[watchdog = std::move(watchdog), w = WeakFromThis()](future<bool> f) mutable {
if (auto self = w.lock()) {
return self->OnStart(std::move(watchdog), f.get());
}
return make_ready_future(false);
});
}References
- Reject Obscured Domain & Return Types: Flag and reject auto when it hides StatusOr, domain objects, protobuf messages/fields, or function return types. (link)
There was a problem hiding this comment.
This (auto f as the lambda parameter) is verbatim the approach used in all the other methods in this class, so IMO it would be more confusing to have just this one specify future<bool>.
|
/gcbrun |
|
The zizmor findings are unrelated to this PR: |
|
I'm fixing the tests now - there were many subclasses of |
|
/gcbrun |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #16569 +/- ##
========================================
Coverage 92.37% 92.38%
========================================
Files 2266 2266
Lines 219493 219636 +143
========================================
+ Hits 202765 202902 +137
- Misses 16728 16734 +6 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Another recent PR (#16571) makes unrelated changes and has the same asan-pr and xsan-pr failures as this one, so I think those failures aren't part of this PR. I can investigate further tomorrow though. |
|
/gharun |
This branch was successfully deployed
2 participants
StartWithCorkedInitialMetadata()allows callers that immediately send an initialWrite()after starting a bidirectional stream to save a network send and agrpc::CompletionQueue::Next()call on stream startup.When a stream is started after calling
grpc::ClientContext::set_initial_metadata_corked(true), it does not send the initial metadata until the firstWrite()call, coalescing it with the first message's network write. It also does not bounce a tag for starting the stream through thegrpc::CompletionQueue. RPCs that can tolerate this behavior have fewer thread hops and less contention ongrpc::CompletionQueue::Next().Unfortunately there's no getter for initial metadata in the
grpc::ClientContext(and anyway we don't plumb theClientContextintogoogle::cloud::internal::CompletionQueueImpl). So we have to implement this at a higher level where theClientContextis available.I personally will use this to optimize bidirectional streaming RPCs, so I put it in
AsyncStreamingReadWriteRpcfor the purposes of review. But analogous and mechanical changes could be done inAsyncStreamingReadRpcandAsyncStreamingWriteRpc.See b/573273671