aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/surface/call.cc
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib/surface/call.cc')
-rw-r--r--src/core/lib/surface/call.cc67
1 files changed, 43 insertions, 24 deletions
diff --git a/src/core/lib/surface/call.cc b/src/core/lib/surface/call.cc
index b07c4d6c10..11b438f5dc 100644
--- a/src/core/lib/surface/call.cc
+++ b/src/core/lib/surface/call.cc
@@ -48,6 +48,7 @@
#include "src/core/lib/surface/call_test_only.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/completion_queue.h"
+#include "src/core/lib/surface/server.h"
#include "src/core/lib/surface/validate_metadata.h"
#include "src/core/lib/transport/error_utils.h"
#include "src/core/lib/transport/metadata.h"
@@ -95,7 +96,7 @@ typedef struct batch_control {
grpc_closure start_batch;
grpc_closure finish_batch;
gpr_refcount steps_to_complete;
- grpc_error* batch_error;
+ gpr_atm batch_error;
grpc_transport_stream_op_batch op;
} batch_control;
@@ -202,6 +203,8 @@ struct grpc_call {
} client;
struct {
int* cancelled;
+ // backpointer to owning server if this is a server side call.
+ grpc_server* server;
} server;
} final_op;
grpc_error* status_error;
@@ -311,14 +314,10 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
/* Always support no compression */
GPR_BITSET(&call->encodings_accepted_by_peer, GRPC_MESSAGE_COMPRESS_NONE);
call->is_client = args->server_transport_data == nullptr;
- if (call->is_client) {
- GRPC_STATS_INC_CLIENT_CALLS_CREATED();
- } else {
- GRPC_STATS_INC_SERVER_CALLS_CREATED();
- }
call->stream_op_payload.context = call->context;
grpc_slice path = grpc_empty_slice();
if (call->is_client) {
+ GRPC_STATS_INC_CLIENT_CALLS_CREATED();
GPR_ASSERT(args->add_initial_metadata_count <
MAX_SEND_EXTRA_METADATA_COUNT);
for (i = 0; i < args->add_initial_metadata_count; i++) {
@@ -332,6 +331,8 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
call->send_extra_metadata_count =
static_cast<int>(args->add_initial_metadata_count);
} else {
+ GRPC_STATS_INC_SERVER_CALLS_CREATED();
+ call->final_op.server.server = args->server;
GPR_ASSERT(args->add_initial_metadata_count == 0);
call->send_extra_metadata_count = 0;
}
@@ -435,10 +436,18 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
&call->pollent);
}
- grpc_core::channelz::ChannelNode* channelz_channel =
- grpc_channel_get_channelz_node(call->channel);
- if (channelz_channel != nullptr) {
- channelz_channel->RecordCallStarted();
+ if (call->is_client) {
+ grpc_core::channelz::ChannelNode* channelz_channel =
+ grpc_channel_get_channelz_node(call->channel);
+ if (channelz_channel != nullptr) {
+ channelz_channel->RecordCallStarted();
+ }
+ } else {
+ grpc_core::channelz::ServerNode* channelz_server =
+ grpc_server_get_channelz_node(call->final_op.server.server);
+ if (channelz_server != nullptr) {
+ channelz_server->RecordCallStarted();
+ }
}
grpc_slice_unref_internal(path);
@@ -709,14 +718,15 @@ static void set_final_status(grpc_call* call, grpc_error* error) {
} else {
*call->final_op.server.cancelled =
error != GRPC_ERROR_NONE || call->status_error != GRPC_ERROR_NONE;
- /* TODO(ncteisen) : Update channelz handling for server
- if (channelz_channel != nullptr) {
+ grpc_core::channelz::ServerNode* channelz_server =
+ grpc_server_get_channelz_node(call->final_op.server.server);
+ if (channelz_server != nullptr) {
if (*call->final_op.server.cancelled) {
- channelz_channel->RecordCallFailed();
+ channelz_server->RecordCallFailed();
} else {
- channelz_channel->RecordCallSucceeded();
+ channelz_server->RecordCallSucceeded();
}
- } */
+ }
GRPC_ERROR_UNREF(error);
}
}
@@ -1106,14 +1116,17 @@ static void finish_batch_completion(void* user_data,
}
static void reset_batch_errors(batch_control* bctl) {
- GRPC_ERROR_UNREF(bctl->batch_error);
- bctl->batch_error = GRPC_ERROR_NONE;
+ GRPC_ERROR_UNREF(
+ reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)));
+ gpr_atm_rel_store(&bctl->batch_error,
+ reinterpret_cast<gpr_atm>(GRPC_ERROR_NONE));
}
static void post_batch_completion(batch_control* bctl) {
grpc_call* next_child_call;
grpc_call* call = bctl->call;
- grpc_error* error = GRPC_ERROR_REF(bctl->batch_error);
+ grpc_error* error = GRPC_ERROR_REF(
+ reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)));
if (bctl->op.send_initial_metadata) {
grpc_metadata_batch_destroy(
@@ -1277,8 +1290,10 @@ static void receiving_stream_ready(void* bctlp, grpc_error* error) {
grpc_call* call = bctl->call;
if (error != GRPC_ERROR_NONE) {
call->receiving_stream.reset();
- if (bctl->batch_error == GRPC_ERROR_NONE) {
- bctl->batch_error = GRPC_ERROR_REF(error);
+ if (reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)) ==
+ GRPC_ERROR_NONE) {
+ gpr_atm_rel_store(&bctl->batch_error,
+ reinterpret_cast<gpr_atm>(GRPC_ERROR_REF(error)));
}
cancel_with_error(call, GRPC_ERROR_REF(error));
}
@@ -1384,8 +1399,10 @@ static void receiving_initial_metadata_ready(void* bctlp, grpc_error* error) {
call->send_deadline = md->deadline;
}
} else {
- if (bctl->batch_error == GRPC_ERROR_NONE) {
- bctl->batch_error = GRPC_ERROR_REF(error);
+ if (reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)) ==
+ GRPC_ERROR_NONE) {
+ gpr_atm_rel_store(&bctl->batch_error,
+ reinterpret_cast<gpr_atm>(GRPC_ERROR_REF(error)));
}
cancel_with_error(call, GRPC_ERROR_REF(error));
}
@@ -1435,8 +1452,10 @@ static void finish_batch(void* bctlp, grpc_error* error) {
batch_control* bctl = static_cast<batch_control*>(bctlp);
grpc_call* call = bctl->call;
GRPC_CALL_COMBINER_STOP(&call->call_combiner, "on_complete");
- if (bctl->batch_error == GRPC_ERROR_NONE) {
- bctl->batch_error = GRPC_ERROR_REF(error);
+ if (reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)) ==
+ GRPC_ERROR_NONE) {
+ gpr_atm_rel_store(&bctl->batch_error,
+ reinterpret_cast<gpr_atm>(GRPC_ERROR_REF(error)));
}
if (error != GRPC_ERROR_NONE) {
cancel_with_error(call, GRPC_ERROR_REF(error));
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-30 09:58:24 +0000