natas_hw
/
grpc
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

dont cancel if done

This commit is contained in:
Craig Tiller 2024-06-11 12:21:37 -07:00
parent eaa4e67d50
commit 9cbe41cb3a
4 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/core/lib/surface/client_call.cc
View File

@ -338,6 +338,7 @@ void ClientCall::CommitBatch(const grpc_op* ops, size_t nops, void* notify_tag,
[this, out_status, out_status_details, out_error_string, [this, out_status, out_status_details, out_error_string,
out_trailing_metadata]( out_trailing_metadata](
ServerMetadataHandle server_trailing_metadata) { ServerMetadataHandle server_trailing_metadata) {
saw_trailing_metadata_.store(true, std::memory_order_relaxed);
ResetDeadline(); ResetDeadline();
GRPC_TRACE_LOG(call, INFO) GRPC_TRACE_LOG(call, INFO)
<< DebugTag() << "RecvStatusOnClient " << DebugTag() << "RecvStatusOnClient "

6
src/core/lib/surface/client_call.h
View File

@ -82,8 +82,9 @@ class ClientCall final
void InternalUnref(const char*) override { WeakUnref(); } void InternalUnref(const char*) override { WeakUnref(); }
void Orphaned() override { void Orphaned() override {
// TODO(ctiller): only when we're not already finished if (!saw_trailing_metadata_.load(std::memory_order_relaxed)) {
CancelWithError(absl::CancelledError()); CancelWithError(absl::CancelledError());
}
} }
void SetCompletionQueue(grpc_completion_queue*) override { void SetCompletionQueue(grpc_completion_queue*) override {
@ -164,6 +165,7 @@ class ClientCall final
ServerMetadataHandle received_initial_metadata_; ServerMetadataHandle received_initial_metadata_;
ServerMetadataHandle received_trailing_metadata_; ServerMetadataHandle received_trailing_metadata_;
bool is_trailers_only_; bool is_trailers_only_;
std::atomic<bool> saw_trailing_metadata_{false};
}; };
grpc_call* MakeClientCall( grpc_call* MakeClientCall(

2
src/core/lib/surface/server_call.cc
View File

@ -188,6 +188,8 @@ void ServerCall::CommitBatch(const grpc_op* ops, size_t nops, void* notify_tag,
[this, cancelled = op->data.recv_close_on_server.cancelled]() { [this, cancelled = op->data.recv_close_on_server.cancelled]() {
return Map(call_handler_.WasCancelled(), return Map(call_handler_.WasCancelled(),
[cancelled, this](bool result) -> Success { [cancelled, this](bool result) -> Success {
saw_was_cancelled_.store(true,
std::memory_order_relaxed);
ResetDeadline(); ResetDeadline();
*cancelled = result ? 1 : 0; *cancelled = result ? 1 : 0;
return Success{}; return Success{};

7
src/core/lib/surface/server_call.h
View File

@ -20,6 +20,7 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <atomic>
#include <cstdint> #include <cstdint>
#include <memory> #include <memory>
#include <string> #include <string>
@ -101,8 +102,9 @@ class ServerCall final : public Call, public DualRefCounted<ServerCall> {
void InternalUnref(const char*) override { WeakUnref(); } void InternalUnref(const char*) override { WeakUnref(); }
void Orphaned() override { void Orphaned() override {
// TODO(ctiller): only when we're not already finished if (!saw_was_cancelled_.load(std::memory_order_relaxed)) {
CancelWithError(absl::CancelledError()); CancelWithError(absl::CancelledError());
}
} }
void SetCompletionQueue(grpc_completion_queue*) override { void SetCompletionQueue(grpc_completion_queue*) override {
@ -155,6 +157,7 @@ class ServerCall final : public Call, public DualRefCounted<ServerCall> {
ClientMetadataHandle client_initial_metadata_stored_; ClientMetadataHandle client_initial_metadata_stored_;
grpc_completion_queue* const cq_; grpc_completion_queue* const cq_;
ServerInterface* const server_; ServerInterface* const server_;
std::atomic<bool> saw_was_cancelled_{false};
}; };
grpc_call* MakeServerCall(CallHandler call_handler, grpc_call* MakeServerCall(CallHandler call_handler,