aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/ext
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/ext')
-rw-r--r--src/core/ext/census/grpc_filter.c14
-rw-r--r--src/core/ext/client_config/client_channel.c6
-rw-r--r--src/core/ext/client_config/subchannel.c25
-rw-r--r--src/core/ext/client_config/subchannel.h4
-rw-r--r--src/core/ext/client_config/subchannel_call_holder.c38
-rw-r--r--src/core/ext/load_reporting/load_reporting_filter.c6
6 files changed, 63 insertions, 30 deletions
diff --git a/src/core/ext/census/grpc_filter.c b/src/core/ext/census/grpc_filter.c
index f51d850e01..55046f4df8 100644
--- a/src/core/ext/census/grpc_filter.c
+++ b/src/core/ext/census/grpc_filter.c
@@ -127,13 +127,14 @@ static void server_start_transport_op(grpc_exec_ctx *exec_ctx,
grpc_call_next_op(exec_ctx, elem, op);
}
-static void client_init_call_elem(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- grpc_call_element_args *args) {
+static grpc_error *client_init_call_elem(grpc_exec_ctx *exec_ctx,
+ grpc_call_element *elem,
+ grpc_call_element_args *args) {
call_data *d = elem->call_data;
GPR_ASSERT(d != NULL);
memset(d, 0, sizeof(*d));
d->start_ts = gpr_now(GPR_CLOCK_REALTIME);
+ return GRPC_ERROR_NONE;
}
static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx,
@@ -145,15 +146,16 @@ static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx,
/* TODO(hongyu): record rpc client stats and census_rpc_end_op here */
}
-static void server_init_call_elem(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- grpc_call_element_args *args) {
+static grpc_error *server_init_call_elem(grpc_exec_ctx *exec_ctx,
+ grpc_call_element *elem,
+ grpc_call_element_args *args) {
call_data *d = elem->call_data;
GPR_ASSERT(d != NULL);
memset(d, 0, sizeof(*d));
d->start_ts = gpr_now(GPR_CLOCK_REALTIME);
/* TODO(hongyu): call census_tracing_start_op here. */
grpc_closure_init(&d->finish_recv, server_on_done_recv, elem);
+ return GRPC_ERROR_NONE;
}
static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c
index a096435c98..b7cff88a8c 100644
--- a/src/core/ext/client_config/client_channel.c
+++ b/src/core/ext/client_config/client_channel.c
@@ -436,10 +436,12 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
}
/* Constructor for call_data */
-static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
- grpc_call_element_args *args) {
+static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
+ grpc_call_element *elem,
+ grpc_call_element_args *args) {
grpc_subchannel_call_holder_init(elem->call_data, cc_pick_subchannel, elem,
args->call_stack);
+ return GRPC_ERROR_NONE;
}
/* Destructor for call_data */
diff --git a/src/core/ext/client_config/subchannel.c b/src/core/ext/client_config/subchannel.c
index d089cd4399..df35904b85 100644
--- a/src/core/ext/client_config/subchannel.c
+++ b/src/core/ext/client_config/subchannel.c
@@ -702,19 +702,26 @@ grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
return GET_CONNECTED_SUBCHANNEL(c, acq);
}
-grpc_subchannel_call *grpc_connected_subchannel_create_call(
+grpc_error *grpc_connected_subchannel_create_call(
grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
- grpc_polling_entity *pollent) {
+ grpc_polling_entity *pollent, grpc_subchannel_call **call) {
grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
- grpc_subchannel_call *call =
- gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
- grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call);
- call->connection = con;
+ *call = gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
+ grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(*call);
+ (*call)->connection = con; // Ref is added below.
+ grpc_error *error =
+ grpc_call_stack_init(exec_ctx, chanstk, 1, subchannel_call_destroy, *call,
+ NULL, NULL, callstk);
+ if (error != GRPC_ERROR_NONE) {
+ const char *error_string = grpc_error_string(error);
+ gpr_log(GPR_ERROR, "error: %s", error_string);
+ grpc_error_free_string(error_string);
+ gpr_free(*call);
+ return error;
+ }
GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call");
- grpc_call_stack_init(exec_ctx, chanstk, 1, subchannel_call_destroy, call,
- NULL, NULL, callstk);
grpc_call_stack_set_pollset_or_pollset_set(exec_ctx, callstk, pollent);
- return call;
+ return GRPC_ERROR_NONE;
}
grpc_call_stack *grpc_subchannel_call_get_call_stack(
diff --git a/src/core/ext/client_config/subchannel.h b/src/core/ext/client_config/subchannel.h
index b6d39f5dc5..ae1d96e640 100644
--- a/src/core/ext/client_config/subchannel.h
+++ b/src/core/ext/client_config/subchannel.h
@@ -108,9 +108,9 @@ void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
/** construct a subchannel call */
-grpc_subchannel_call *grpc_connected_subchannel_create_call(
+grpc_error *grpc_connected_subchannel_create_call(
grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *connected_subchannel,
- grpc_polling_entity *pollent);
+ grpc_polling_entity *pollent, grpc_subchannel_call **subchannel_call);
/** process a transport level op */
void grpc_connected_subchannel_process_transport_op(
diff --git a/src/core/ext/client_config/subchannel_call_holder.c b/src/core/ext/client_config/subchannel_call_holder.c
index b96a0ad093..ac286825f6 100644
--- a/src/core/ext/client_config/subchannel_call_holder.c
+++ b/src/core/ext/client_config/subchannel_call_holder.c
@@ -84,6 +84,11 @@ void grpc_subchannel_call_holder_destroy(grpc_exec_ctx *exec_ctx,
gpr_free(holder->waiting_ops);
}
+// The logic here is fairly complicated, due to (a) the fact that we
+// need to handle the case where we receive the send op before the
+// initial metadata op, and (b) the need for efficiency, especially in
+// the streaming case.
+// TODO(ctiller): Explain this more thoroughly.
void grpc_subchannel_call_holder_perform_op(grpc_exec_ctx *exec_ctx,
grpc_subchannel_call_holder *holder,
grpc_transport_stream_op *op) {
@@ -121,7 +126,8 @@ retry:
}
/* if this is a cancellation, then we can raise our cancelled flag */
if (op->cancel_error != GRPC_ERROR_NONE) {
- if (!gpr_atm_rel_cas(&holder->subchannel_call, 0, 1)) {
+ if (!gpr_atm_rel_cas(&holder->subchannel_call, 0,
+ (gpr_atm)(uintptr_t)CANCELLED_CALL)) {
goto retry;
} else {
switch (holder->creation_phase) {
@@ -158,10 +164,17 @@ retry:
/* if we've got a subchannel, then let's ask it to create a call */
if (holder->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING &&
holder->connected_subchannel != NULL) {
- gpr_atm_rel_store(
- &holder->subchannel_call,
- (gpr_atm)(uintptr_t)grpc_connected_subchannel_create_call(
- exec_ctx, holder->connected_subchannel, holder->pollent));
+ grpc_subchannel_call *subchannel_call = NULL;
+ grpc_error *error = grpc_connected_subchannel_create_call(
+ exec_ctx, holder->connected_subchannel, holder->pollent,
+ &subchannel_call);
+ if (error != GRPC_ERROR_NONE) {
+ subchannel_call = CANCELLED_CALL;
+ fail_locked(exec_ctx, holder, error);
+ grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error);
+ }
+ gpr_atm_rel_store(&holder->subchannel_call,
+ (gpr_atm)(uintptr_t)subchannel_call);
retry_waiting_locked(exec_ctx, holder);
goto retry;
}
@@ -189,10 +202,17 @@ static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg,
GRPC_ERROR_CREATE_REFERENCING(
"Cancelled before creating subchannel", &error, 1));
} else {
- gpr_atm_rel_store(
- &holder->subchannel_call,
- (gpr_atm)(uintptr_t)grpc_connected_subchannel_create_call(
- exec_ctx, holder->connected_subchannel, holder->pollent));
+ grpc_subchannel_call *subchannel_call = NULL;
+ grpc_error *new_error = grpc_connected_subchannel_create_call(
+ exec_ctx, holder->connected_subchannel, holder->pollent,
+ &subchannel_call);
+ if (new_error != GRPC_ERROR_NONE) {
+ new_error = grpc_error_add_child(new_error, error);
+ subchannel_call = CANCELLED_CALL;
+ fail_locked(exec_ctx, holder, new_error);
+ }
+ gpr_atm_rel_store(&holder->subchannel_call,
+ (gpr_atm)(uintptr_t)subchannel_call);
retry_waiting_locked(exec_ctx, holder);
}
gpr_mu_unlock(&holder->mu);
diff --git a/src/core/ext/load_reporting/load_reporting_filter.c b/src/core/ext/load_reporting/load_reporting_filter.c
index f372f88c3a..b584e31c5d 100644
--- a/src/core/ext/load_reporting/load_reporting_filter.c
+++ b/src/core/ext/load_reporting/load_reporting_filter.c
@@ -56,10 +56,12 @@ static void invoke_lr_fn_locked(grpc_load_reporting_config *lrc,
}
/* Constructor for call_data */
-static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
- grpc_call_element_args *args) {
+static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
+ grpc_call_element *elem,
+ grpc_call_element_args *args) {
call_data *calld = elem->call_data;
memset(calld, 0, sizeof(call_data));
+ return GRPC_ERROR_NONE;
}
/* Destructor for call_data */
generated by cgit v1.2.3 (git 2.39.1) at 2025-02-01 05:27:43 +0000