aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/ext/client_channel
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/ext/client_channel')
-rw-r--r--src/core/ext/client_channel/channel_connectivity.c8
-rw-r--r--src/core/ext/client_channel/client_channel.c78
-rw-r--r--src/core/ext/client_channel/client_channel_factory.c6
-rw-r--r--src/core/ext/client_channel/client_channel_plugin.c11
-rw-r--r--src/core/ext/client_channel/http_connect_handshaker.c27
-rw-r--r--src/core/ext/client_channel/lb_policy_factory.c10
-rw-r--r--src/core/ext/client_channel/lb_policy_factory.h5
-rw-r--r--src/core/ext/client_channel/subchannel.c30
-rw-r--r--src/core/ext/client_channel/subchannel_index.c2
-rw-r--r--src/core/ext/client_channel/uri_parser.c44
10 files changed, 119 insertions, 102 deletions
diff --git a/src/core/ext/client_channel/channel_connectivity.c b/src/core/ext/client_channel/channel_connectivity.c
index 9797e66564..dd70bc2c6c 100644
--- a/src/core/ext/client_channel/channel_connectivity.c
+++ b/src/core/ext/client_channel/channel_connectivity.c
@@ -76,6 +76,7 @@ typedef struct {
gpr_mu mu;
callback_phase phase;
grpc_closure on_complete;
+ grpc_closure on_timeout;
grpc_timer alarm;
grpc_connectivity_state state;
grpc_completion_queue *cq;
@@ -198,7 +199,10 @@ void grpc_channel_watch_connectivity_state(
grpc_cq_begin_op(cq, tag);
gpr_mu_init(&w->mu);
- grpc_closure_init(&w->on_complete, watch_complete, w);
+ grpc_closure_init(&w->on_complete, watch_complete, w,
+ grpc_schedule_on_exec_ctx);
+ grpc_closure_init(&w->on_timeout, timeout_complete, w,
+ grpc_schedule_on_exec_ctx);
w->phase = WAITING;
w->state = last_observed_state;
w->cq = cq;
@@ -207,7 +211,7 @@ void grpc_channel_watch_connectivity_state(
grpc_timer_init(&exec_ctx, &w->alarm,
gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
- timeout_complete, w, gpr_now(GPR_CLOCK_MONOTONIC));
+ &w->on_timeout, gpr_now(GPR_CLOCK_MONOTONIC));
if (client_channel_elem->filter == &grpc_client_channel_filter) {
GRPC_CHANNEL_INTERNAL_REF(channel, "watch_channel_connectivity");
diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c
index 9d46338428..2f25fef9a7 100644
--- a/src/core/ext/client_channel/client_channel.c
+++ b/src/core/ext/client_channel/client_channel.c
@@ -83,8 +83,12 @@ static void *method_parameters_copy(void *value) {
return new_value;
}
+static void method_parameters_free(grpc_exec_ctx *exec_ctx, void *p) {
+ gpr_free(p);
+}
+
static const grpc_mdstr_hash_table_vtable method_parameters_vtable = {
- gpr_free, method_parameters_copy};
+ method_parameters_free, method_parameters_copy};
static void *method_parameters_create_from_json(const grpc_json *json) {
wait_for_ready_value wait_for_ready = WAIT_FOR_READY_UNSET;
@@ -249,7 +253,8 @@ static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand,
GRPC_CHANNEL_STACK_REF(chand->owning_stack, "watch_lb_policy");
w->chand = chand;
- grpc_closure_init(&w->on_changed, on_lb_policy_state_changed, w);
+ grpc_closure_init(&w->on_changed, on_lb_policy_state_changed, w,
+ grpc_schedule_on_exec_ctx);
w->state = current_state;
w->lb_policy = lb_policy;
grpc_lb_policy_notify_on_state_change(exec_ctx, lb_policy, &w->state,
@@ -327,7 +332,7 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
grpc_service_config_create(service_config_json);
if (service_config != NULL) {
method_params_table = grpc_service_config_create_method_config_table(
- service_config, method_parameters_create_from_json,
+ exec_ctx, service_config, method_parameters_create_from_json,
&method_parameters_vtable);
grpc_service_config_destroy(service_config);
}
@@ -336,7 +341,7 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
// be pointing to data inside chand->resolver_result.
// The copy will be saved in chand->lb_policy_name below.
lb_policy_name = gpr_strdup(lb_policy_name);
- grpc_channel_args_destroy(chand->resolver_result);
+ grpc_channel_args_destroy(exec_ctx, chand->resolver_result);
chand->resolver_result = NULL;
}
@@ -357,18 +362,16 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
chand->service_config_json = service_config_json;
}
if (chand->method_params_table != NULL) {
- grpc_mdstr_hash_table_unref(chand->method_params_table);
+ grpc_mdstr_hash_table_unref(exec_ctx, chand->method_params_table);
}
chand->method_params_table = method_params_table;
if (lb_policy != NULL) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures,
- NULL);
+ grpc_closure_list_sched(exec_ctx, &chand->waiting_for_config_closures);
} else if (chand->resolver == NULL /* disconnected */) {
grpc_closure_list_fail_all(
&chand->waiting_for_config_closures,
GRPC_ERROR_CREATE_REFERENCING("Channel disconnected", &error, 1));
- grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures,
- NULL);
+ grpc_closure_list_sched(exec_ctx, &chand->waiting_for_config_closures);
}
if (lb_policy != NULL && chand->exit_idle_when_lb_policy_arrives) {
GRPC_LB_POLICY_REF(lb_policy, "exit_idle");
@@ -425,7 +428,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
grpc_transport_op *op) {
channel_data *chand = elem->channel_data;
- grpc_exec_ctx_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE, NULL);
+ grpc_closure_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE);
GPR_ASSERT(op->set_accept_stream == false);
if (op->bind_pollset != NULL) {
@@ -444,9 +447,8 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
if (op->send_ping != NULL) {
if (chand->lb_policy == NULL) {
- grpc_exec_ctx_sched(exec_ctx, op->send_ping,
- GRPC_ERROR_CREATE("Ping with no load balancing"),
- NULL);
+ grpc_closure_sched(exec_ctx, op->send_ping,
+ GRPC_ERROR_CREATE("Ping with no load balancing"));
} else {
grpc_lb_policy_ping_one(exec_ctx, chand->lb_policy, op->send_ping);
op->bind_pollset = NULL;
@@ -465,8 +467,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
if (!chand->started_resolving) {
grpc_closure_list_fail_all(&chand->waiting_for_config_closures,
GRPC_ERROR_REF(op->disconnect_with_error));
- grpc_exec_ctx_enqueue_list(exec_ctx,
- &chand->waiting_for_config_closures, NULL);
+ grpc_closure_list_sched(exec_ctx, &chand->waiting_for_config_closures);
}
if (chand->lb_policy != NULL) {
grpc_pollset_set_del_pollset_set(exec_ctx,
@@ -511,7 +512,8 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
gpr_mu_init(&chand->mu);
chand->owning_stack = args->channel_stack;
grpc_closure_init(&chand->on_resolver_result_changed,
- on_resolver_result_changed, chand);
+ on_resolver_result_changed, chand,
+ grpc_schedule_on_exec_ctx);
chand->interested_parties = grpc_pollset_set_create();
grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE,
"client_channel");
@@ -556,7 +558,7 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
gpr_free(chand->lb_policy_name);
gpr_free(chand->service_config_json);
if (chand->method_params_table != NULL) {
- grpc_mdstr_hash_table_unref(chand->method_params_table);
+ grpc_mdstr_hash_table_unref(exec_ctx, chand->method_params_table);
}
grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
grpc_pollset_set_destroy(chand->interested_parties);
@@ -678,8 +680,9 @@ static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
calld->waiting_ops_count = 0;
calld->waiting_ops_capacity = 0;
GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops");
- grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(retry_ops, a),
- GRPC_ERROR_NONE, NULL);
+ grpc_closure_sched(
+ exec_ctx, grpc_closure_create(retry_ops, a, grpc_schedule_on_exec_ctx),
+ GRPC_ERROR_NONE);
}
static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg,
@@ -761,14 +764,14 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg,
if (cpa->connected_subchannel == NULL) {
/* cancelled, do nothing */
} else if (error != GRPC_ERROR_NONE) {
- grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error), NULL);
+ grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error));
} else {
call_data *calld = cpa->elem->call_data;
gpr_mu_lock(&calld->mu);
if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
cpa->initial_metadata_flags, cpa->connected_subchannel,
cpa->on_ready, GRPC_ERROR_NONE)) {
- grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL);
+ grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE);
}
gpr_mu_unlock(&calld->mu);
}
@@ -800,9 +803,9 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
cpa = closure->cb_arg;
if (cpa->connected_subchannel == connected_subchannel) {
cpa->connected_subchannel = NULL;
- grpc_exec_ctx_sched(
+ grpc_closure_sched(
exec_ctx, cpa->on_ready,
- GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1), NULL);
+ GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1));
}
}
gpr_mu_unlock(&chand->mu);
@@ -853,12 +856,12 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
cpa->connected_subchannel = connected_subchannel;
cpa->on_ready = on_ready;
cpa->elem = elem;
- grpc_closure_init(&cpa->closure, continue_picking, cpa);
+ grpc_closure_init(&cpa->closure, continue_picking, cpa,
+ grpc_schedule_on_exec_ctx);
grpc_closure_list_append(&chand->waiting_for_config_closures, &cpa->closure,
GRPC_ERROR_NONE);
} else {
- grpc_exec_ctx_sched(exec_ctx, on_ready, GRPC_ERROR_CREATE("Disconnected"),
- NULL);
+ grpc_closure_sched(exec_ctx, on_ready, GRPC_ERROR_CREATE("Disconnected"));
}
gpr_mu_unlock(&chand->mu);
@@ -943,7 +946,8 @@ retry:
calld->connected_subchannel == NULL &&
op->send_initial_metadata != NULL) {
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL;
- grpc_closure_init(&calld->next_step, subchannel_ready, elem);
+ grpc_closure_init(&calld->next_step, subchannel_ready, elem,
+ grpc_schedule_on_exec_ctx);
GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel");
/* If a subchannel is not available immediately, the polling entity from
call_data should be provided to channel_data's interested_parties, so
@@ -1001,8 +1005,8 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
gpr_mu_unlock(&chand->mu);
// If the method config table was present, use it.
if (method_params_table != NULL) {
- const method_parameters *method_params =
- grpc_method_config_table_get(method_params_table, calld->path);
+ const method_parameters *method_params = grpc_method_config_table_get(
+ exec_ctx, method_params_table, calld->path);
if (method_params != NULL) {
const bool have_method_timeout =
gpr_time_cmp(method_params->timeout, gpr_time_0(GPR_TIMESPAN)) != 0;
@@ -1025,7 +1029,7 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
gpr_mu_unlock(&calld->mu);
}
}
- grpc_mdstr_hash_table_unref(method_params_table);
+ grpc_mdstr_hash_table_unref(exec_ctx, method_params_table);
}
}
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config");
@@ -1066,8 +1070,8 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
grpc_mdstr_hash_table *method_params_table =
grpc_mdstr_hash_table_ref(chand->method_params_table);
gpr_mu_unlock(&chand->mu);
- method_parameters *method_params =
- grpc_method_config_table_get(method_params_table, args->path);
+ method_parameters *method_params = grpc_method_config_table_get(
+ exec_ctx, method_params_table, args->path);
if (method_params != NULL) {
if (gpr_time_cmp(method_params->timeout,
gpr_time_0(GPR_CLOCK_MONOTONIC)) != 0) {
@@ -1080,7 +1084,7 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
method_params->wait_for_ready;
}
}
- grpc_mdstr_hash_table_unref(method_params_table);
+ grpc_mdstr_hash_table_unref(exec_ctx, method_params_table);
} else {
gpr_mu_unlock(&chand->mu);
}
@@ -1089,7 +1093,8 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
// get the service config data once the resolver returns.
// Take a reference to the call stack to be owned by the callback.
GRPC_CALL_STACK_REF(calld->owning_call, "read_service_config");
- grpc_closure_init(&calld->read_service_config, read_service_config, elem);
+ grpc_closure_init(&calld->read_service_config, read_service_config, elem,
+ grpc_schedule_on_exec_ctx);
grpc_closure_list_append(&chand->waiting_for_config_closures,
&calld->read_service_config, GRPC_ERROR_NONE);
gpr_mu_unlock(&chand->mu);
@@ -1108,7 +1113,7 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
void *and_free_memory) {
call_data *calld = elem->call_data;
grpc_deadline_state_destroy(exec_ctx, elem);
- GRPC_MDSTR_UNREF(calld->path);
+ GRPC_MDSTR_UNREF(exec_ctx, calld->path);
GRPC_ERROR_UNREF(calld->cancel_error);
grpc_subchannel_call *call = GET_CALL(calld);
if (call != NULL && call != CANCELLED_CALL) {
@@ -1202,7 +1207,8 @@ void grpc_client_channel_watch_connectivity_state(
w->pollset = pollset;
w->on_complete = on_complete;
grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties, pollset);
- grpc_closure_init(&w->my_closure, on_external_watch_complete, w);
+ grpc_closure_init(&w->my_closure, on_external_watch_complete, w,
+ grpc_schedule_on_exec_ctx);
GRPC_CHANNEL_STACK_REF(w->chand->owning_stack,
"external_connectivity_watcher");
gpr_mu_lock(&chand->mu);
diff --git a/src/core/ext/client_channel/client_channel_factory.c b/src/core/ext/client_channel/client_channel_factory.c
index 4eb35dfcf7..d2707a1556 100644
--- a/src/core/ext/client_channel/client_channel_factory.c
+++ b/src/core/ext/client_channel/client_channel_factory.c
@@ -61,12 +61,10 @@ static void* factory_arg_copy(void* factory) {
return factory;
}
-static void factory_arg_destroy(void* factory) {
+static void factory_arg_destroy(grpc_exec_ctx* exec_ctx, void* factory) {
// TODO(roth): Remove local exec_ctx when
// https://github.com/grpc/grpc/pull/8705 is merged.
- grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- grpc_client_channel_factory_unref(&exec_ctx, factory);
- grpc_exec_ctx_finish(&exec_ctx);
+ grpc_client_channel_factory_unref(exec_ctx, factory);
}
static int factory_arg_cmp(void* factory1, void* factory2) {
diff --git a/src/core/ext/client_channel/client_channel_plugin.c b/src/core/ext/client_channel/client_channel_plugin.c
index a3e5079843..988b7a1d5c 100644
--- a/src/core/ext/client_channel/client_channel_plugin.c
+++ b/src/core/ext/client_channel/client_channel_plugin.c
@@ -43,12 +43,14 @@
#include "src/core/ext/client_channel/subchannel_index.h"
#include "src/core/lib/surface/channel_init.h"
-static bool append_filter(grpc_channel_stack_builder *builder, void *arg) {
+static bool append_filter(grpc_exec_ctx *exec_ctx,
+ grpc_channel_stack_builder *builder, void *arg) {
return grpc_channel_stack_builder_append_filter(
builder, (const grpc_channel_filter *)arg, NULL, NULL);
}
-static bool set_default_host_if_unset(grpc_channel_stack_builder *builder,
+static bool set_default_host_if_unset(grpc_exec_ctx *exec_ctx,
+ grpc_channel_stack_builder *builder,
void *unused) {
const grpc_channel_args *args =
grpc_channel_stack_builder_get_channel_arguments(builder);
@@ -66,9 +68,10 @@ static bool set_default_host_if_unset(grpc_channel_stack_builder *builder,
arg.key = GRPC_ARG_DEFAULT_AUTHORITY;
arg.value.string = default_authority;
grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1);
- grpc_channel_stack_builder_set_channel_arguments(builder, new_args);
+ grpc_channel_stack_builder_set_channel_arguments(exec_ctx, builder,
+ new_args);
gpr_free(default_authority);
- grpc_channel_args_destroy(new_args);
+ grpc_channel_args_destroy(exec_ctx, new_args);
}
return true;
}
diff --git a/src/core/ext/client_channel/http_connect_handshaker.c b/src/core/ext/client_channel/http_connect_handshaker.c
index 76c78ee853..27b117af84 100644
--- a/src/core/ext/client_channel/http_connect_handshaker.c
+++ b/src/core/ext/client_channel/http_connect_handshaker.c
@@ -46,6 +46,7 @@
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/http/format_request.h"
#include "src/core/lib/http/parser.h"
+#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/support/env.h"
typedef struct http_connect_handshaker {
@@ -83,11 +84,12 @@ static void http_connect_handshaker_unref(grpc_exec_ctx* exec_ctx,
grpc_endpoint_destroy(exec_ctx, handshaker->endpoint_to_destroy);
}
if (handshaker->read_buffer_to_destroy != NULL) {
- grpc_slice_buffer_destroy(handshaker->read_buffer_to_destroy);
+ grpc_slice_buffer_destroy_internal(exec_ctx,
+ handshaker->read_buffer_to_destroy);
gpr_free(handshaker->read_buffer_to_destroy);
}
gpr_free(handshaker->proxy_server);
- grpc_slice_buffer_destroy(&handshaker->write_buffer);
+ grpc_slice_buffer_destroy_internal(exec_ctx, &handshaker->write_buffer);
grpc_http_parser_destroy(&handshaker->http_parser);
grpc_http_response_destroy(&handshaker->http_response);
gpr_free(handshaker);
@@ -97,12 +99,12 @@ static void http_connect_handshaker_unref(grpc_exec_ctx* exec_ctx,
// Set args fields to NULL, saving the endpoint and read buffer for
// later destruction.
static void cleanup_args_for_failure_locked(
- http_connect_handshaker* handshaker) {
+ grpc_exec_ctx* exec_ctx, http_connect_handshaker* handshaker) {
handshaker->endpoint_to_destroy = handshaker->args->endpoint;
handshaker->args->endpoint = NULL;
handshaker->read_buffer_to_destroy = handshaker->args->read_buffer;
handshaker->args->read_buffer = NULL;
- grpc_channel_args_destroy(handshaker->args->args);
+ grpc_channel_args_destroy(exec_ctx, handshaker->args->args);
handshaker->args->args = NULL;
}
@@ -125,13 +127,13 @@ static void handshake_failed_locked(grpc_exec_ctx* exec_ctx,
grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint);
// Not shutting down, so the handshake failed. Clean up before
// invoking the callback.
- cleanup_args_for_failure_locked(handshaker);
+ cleanup_args_for_failure_locked(exec_ctx, handshaker);
// Set shutdown to true so that subsequent calls to
// http_connect_handshaker_shutdown() do nothing.
handshaker->shutdown = true;
}
// Invoke callback.
- grpc_exec_ctx_sched(exec_ctx, handshaker->on_handshake_done, error, NULL);
+ grpc_closure_sched(exec_ctx, handshaker->on_handshake_done, error);
}
// Callback invoked when finished writing HTTP CONNECT request.
@@ -193,7 +195,7 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
&handshaker->args->read_buffer->slices[i + 1],
handshaker->args->read_buffer->count - i - 1);
grpc_slice_buffer_swap(handshaker->args->read_buffer, &tmp_buffer);
- grpc_slice_buffer_destroy(&tmp_buffer);
+ grpc_slice_buffer_destroy_internal(exec_ctx, &tmp_buffer);
break;
}
}
@@ -210,7 +212,8 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
// complete (e.g., handling chunked transfer encoding or looking
// at the Content-Length: header).
if (handshaker->http_parser.state != GRPC_HTTP_BODY) {
- grpc_slice_buffer_reset_and_unref(handshaker->args->read_buffer);
+ grpc_slice_buffer_reset_and_unref_internal(exec_ctx,
+ handshaker->args->read_buffer);
grpc_endpoint_read(exec_ctx, handshaker->args->endpoint,
handshaker->args->read_buffer,
&handshaker->response_read_closure);
@@ -229,7 +232,7 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
goto done;
}
// Success. Invoke handshake-done callback.
- grpc_exec_ctx_sched(exec_ctx, handshaker->on_handshake_done, error, NULL);
+ grpc_closure_sched(exec_ctx, handshaker->on_handshake_done, error);
done:
// Set shutdown to true so that subsequent calls to
// http_connect_handshaker_shutdown() do nothing.
@@ -255,7 +258,7 @@ static void http_connect_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
if (!handshaker->shutdown) {
handshaker->shutdown = true;
grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint);
- cleanup_args_for_failure_locked(handshaker);
+ cleanup_args_for_failure_locked(exec_ctx, handshaker);
}
gpr_mu_unlock(&handshaker->mu);
}
@@ -313,9 +316,9 @@ grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server) {
handshaker->proxy_server = gpr_strdup(proxy_server);
grpc_slice_buffer_init(&handshaker->write_buffer);
grpc_closure_init(&handshaker->request_done_closure, on_write_done,
- handshaker);
+ handshaker, grpc_schedule_on_exec_ctx);
grpc_closure_init(&handshaker->response_read_closure, on_read_done,
- handshaker);
+ handshaker, grpc_schedule_on_exec_ctx);
grpc_http_parser_init(&handshaker->http_parser, GRPC_HTTP_RESPONSE,
&handshaker->http_response);
return &handshaker->base;
diff --git a/src/core/ext/client_channel/lb_policy_factory.c b/src/core/ext/client_channel/lb_policy_factory.c
index 8a474c8818..7af9bb0411 100644
--- a/src/core/ext/client_channel/lb_policy_factory.c
+++ b/src/core/ext/client_channel/lb_policy_factory.c
@@ -112,11 +112,13 @@ int grpc_lb_addresses_cmp(const grpc_lb_addresses* addresses1,
return 0;
}
-void grpc_lb_addresses_destroy(grpc_lb_addresses* addresses) {
+void grpc_lb_addresses_destroy(grpc_exec_ctx* exec_ctx,
+ grpc_lb_addresses* addresses) {
for (size_t i = 0; i < addresses->num_addresses; ++i) {
gpr_free(addresses->addresses[i].balancer_name);
if (addresses->addresses[i].user_data != NULL) {
- addresses->user_data_vtable->destroy(addresses->addresses[i].user_data);
+ addresses->user_data_vtable->destroy(exec_ctx,
+ addresses->addresses[i].user_data);
}
}
gpr_free(addresses->addresses);
@@ -126,8 +128,8 @@ void grpc_lb_addresses_destroy(grpc_lb_addresses* addresses) {
static void* lb_addresses_copy(void* addresses) {
return grpc_lb_addresses_copy(addresses);
}
-static void lb_addresses_destroy(void* addresses) {
- grpc_lb_addresses_destroy(addresses);
+static void lb_addresses_destroy(grpc_exec_ctx* exec_ctx, void* addresses) {
+ grpc_lb_addresses_destroy(exec_ctx, addresses);
}
static int lb_addresses_cmp(void* addresses1, void* addresses2) {
return grpc_lb_addresses_cmp(addresses1, addresses2);
diff --git a/src/core/ext/client_channel/lb_policy_factory.h b/src/core/ext/client_channel/lb_policy_factory.h
index 79b3dee259..9b8b03f982 100644
--- a/src/core/ext/client_channel/lb_policy_factory.h
+++ b/src/core/ext/client_channel/lb_policy_factory.h
@@ -64,7 +64,7 @@ typedef struct grpc_lb_address {
typedef struct grpc_lb_user_data_vtable {
void *(*copy)(void *);
- void (*destroy)(void *);
+ void (*destroy)(grpc_exec_ctx *exec_ctx, void *);
int (*cmp)(void *, void *);
} grpc_lb_user_data_vtable;
@@ -96,7 +96,8 @@ int grpc_lb_addresses_cmp(const grpc_lb_addresses *addresses1,
const grpc_lb_addresses *addresses2);
/** Destroys \a addresses. */
-void grpc_lb_addresses_destroy(grpc_lb_addresses *addresses);
+void grpc_lb_addresses_destroy(grpc_exec_ctx *exec_ctx,
+ grpc_lb_addresses *addresses);
/** Returns a channel arg containing \a addresses. */
grpc_arg grpc_lb_addresses_create_channel_arg(
diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c
index f294e69392..1bac82b451 100644
--- a/src/core/ext/client_channel/subchannel.c
+++ b/src/core/ext/client_channel/subchannel.c
@@ -46,6 +46,7 @@
#include "src/core/lib/channel/connected_channel.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/support/backoff.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/channel_init.h"
@@ -108,6 +109,9 @@ struct grpc_subchannel {
/** callback for connection finishing */
grpc_closure connected;
+ /** callback for our alarm */
+ grpc_closure on_alarm;
+
/** pollset_set tracking who's interested in a connection
being setup */
grpc_pollset_set *pollset_set;
@@ -206,9 +210,9 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) {
grpc_subchannel *c = arg;
gpr_free((void *)c->filters);
- grpc_channel_args_destroy(c->args);
+ grpc_channel_args_destroy(exec_ctx, c->args);
gpr_free(c->addr);
- grpc_slice_unref(c->initial_connect_string);
+ grpc_slice_unref_internal(exec_ctx, c->initial_connect_string);
grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
grpc_connector_unref(exec_ctx, c->connector);
grpc_pollset_set_destroy(c->pollset_set);
@@ -293,8 +297,9 @@ void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
gpr_atm old_refs;
old_refs = ref_mutate(c, -(gpr_atm)1, 1 REF_MUTATE_PURPOSE("WEAK_UNREF"));
if (old_refs == 1) {
- grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(subchannel_destroy, c),
- GRPC_ERROR_NONE, NULL);
+ grpc_closure_sched(exec_ctx, grpc_closure_create(subchannel_destroy, c,
+ grpc_schedule_on_exec_ctx),
+ GRPC_ERROR_NONE);
}
}
@@ -330,7 +335,8 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
c->args = grpc_channel_args_copy(args->args);
c->root_external_state_watcher.next = c->root_external_state_watcher.prev =
&c->root_external_state_watcher;
- grpc_closure_init(&c->connected, subchannel_connected, c);
+ grpc_closure_init(&c->connected, subchannel_connected, c,
+ grpc_schedule_on_exec_ctx);
grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE,
"subchannel");
int initial_backoff_ms =
@@ -480,7 +486,8 @@ static void maybe_start_connecting_locked(grpc_exec_ctx *exec_ctx,
gpr_log(GPR_INFO, "Retry in %" PRId64 ".%09d seconds",
time_til_next.tv_sec, time_til_next.tv_nsec);
}
- grpc_timer_init(exec_ctx, &c->alarm, c->next_attempt, on_alarm, c, now);
+ grpc_closure_init(&c->on_alarm, on_alarm, c, grpc_schedule_on_exec_ctx);
+ grpc_timer_init(exec_ctx, &c->alarm, c->next_attempt, &c->on_alarm, now);
}
}
@@ -505,7 +512,8 @@ void grpc_subchannel_notify_on_state_change(
w->subchannel = c;
w->pollset_set = interested_parties;
w->notify = notify;
- grpc_closure_init(&w->closure, on_external_state_watcher_done, w);
+ grpc_closure_init(&w->closure, on_external_state_watcher_done, w,
+ grpc_schedule_on_exec_ctx);
if (interested_parties != NULL) {
grpc_pollset_set_add_pollset_set(exec_ctx, c->pollset_set,
interested_parties);
@@ -600,13 +608,13 @@ static void publish_transport_locked(grpc_exec_ctx *exec_ctx,
/* construct channel stack */
grpc_channel_stack_builder *builder = grpc_channel_stack_builder_create();
grpc_channel_stack_builder_set_channel_arguments(
- builder, c->connecting_result.channel_args);
+ exec_ctx, builder, c->connecting_result.channel_args);
grpc_channel_stack_builder_set_transport(builder,
c->connecting_result.transport);
if (!grpc_channel_init_create_stack(exec_ctx, builder,
GRPC_CLIENT_SUBCHANNEL)) {
- grpc_channel_stack_builder_destroy(builder);
+ grpc_channel_stack_builder_destroy(exec_ctx, builder);
abort(); /* TODO(ctiller): what to do here (previously we just crashed) */
}
grpc_error *error = grpc_channel_stack_builder_finish(
@@ -626,7 +634,7 @@ static void publish_transport_locked(grpc_exec_ctx *exec_ctx,
sw_subchannel->subchannel = c;
sw_subchannel->connectivity_state = GRPC_CHANNEL_READY;
grpc_closure_init(&sw_subchannel->closure, subchannel_on_child_state_changed,
- sw_subchannel);
+ sw_subchannel, grpc_schedule_on_exec_ctx);
if (c->disconnected) {
gpr_free(sw_subchannel);
@@ -686,7 +694,7 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg,
}
gpr_mu_unlock(&c->mu);
GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connected");
- grpc_channel_args_destroy(delete_channel_args);
+ grpc_channel_args_destroy(exec_ctx, delete_channel_args);
}
/*
diff --git a/src/core/ext/client_channel/subchannel_index.c b/src/core/ext/client_channel/subchannel_index.c
index a1ba5e945c..1ebe03ef11 100644
--- a/src/core/ext/client_channel/subchannel_index.c
+++ b/src/core/ext/client_channel/subchannel_index.c
@@ -128,7 +128,7 @@ void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx,
grpc_subchannel_key *k) {
grpc_connector_unref(exec_ctx, k->connector);
gpr_free((grpc_channel_args *)k->args.filters);
- grpc_channel_args_destroy((grpc_channel_args *)k->args.args);
+ grpc_channel_args_destroy(exec_ctx, (grpc_channel_args *)k->args.args);
gpr_free(k->args.addr);
gpr_free(k);
}
diff --git a/src/core/ext/client_channel/uri_parser.c b/src/core/ext/client_channel/uri_parser.c
index 0fbc542ef8..f8c946b275 100644
--- a/src/core/ext/client_channel/uri_parser.c
+++ b/src/core/ext/client_channel/uri_parser.c
@@ -42,7 +42,6 @@
#include <grpc/support/port_platform.h>
#include <grpc/support/string_util.h>
-#include "src/core/lib/slice/slice_string_helpers.h"
#include "src/core/lib/support/string.h"
/** a size_t default value... maps to all 1's */
@@ -138,7 +137,6 @@ static int parse_fragment_or_query(const char *uri_text, size_t *i) {
return 1;
}
-static void do_nothing(void *ignored) {}
static void parse_query_parts(grpc_uri *uri) {
static const char *QUERY_PARTS_SEPARATOR = "&";
static const char *QUERY_PARTS_VALUE_SEPARATOR = "=";
@@ -149,38 +147,32 @@ static void parse_query_parts(grpc_uri *uri) {
uri->num_query_parts = 0;
return;
}
- grpc_slice query_slice =
- grpc_slice_new(uri->query, strlen(uri->query), do_nothing);
- grpc_slice_buffer query_parts; /* the &-separated elements of the query */
- grpc_slice_buffer query_param_parts; /* the =-separated subelements */
- grpc_slice_buffer_init(&query_parts);
- grpc_slice_buffer_init(&query_param_parts);
-
- grpc_slice_split(query_slice, QUERY_PARTS_SEPARATOR, &query_parts);
- uri->query_parts = gpr_malloc(query_parts.count * sizeof(char *));
- uri->query_parts_values = gpr_malloc(query_parts.count * sizeof(char *));
- uri->num_query_parts = query_parts.count;
- for (size_t i = 0; i < query_parts.count; i++) {
- grpc_slice_split(query_parts.slices[i], QUERY_PARTS_VALUE_SEPARATOR,
- &query_param_parts);
- GPR_ASSERT(query_param_parts.count > 0);
- uri->query_parts[i] =
- grpc_dump_slice(query_param_parts.slices[0], GPR_DUMP_ASCII);
- if (query_param_parts.count > 1) {
+ gpr_string_split(uri->query, QUERY_PARTS_SEPARATOR, &uri->query_parts,
+ &uri->num_query_parts);
+ uri->query_parts_values = gpr_malloc(uri->num_query_parts * sizeof(char **));
+ for (size_t i = 0; i < uri->num_query_parts; i++) {
+ char **query_param_parts;
+ size_t num_query_param_parts;
+ char *full = uri->query_parts[i];
+ gpr_string_split(full, QUERY_PARTS_VALUE_SEPARATOR, &query_param_parts,
+ &num_query_param_parts);
+ GPR_ASSERT(num_query_param_parts > 0);
+ uri->query_parts[i] = query_param_parts[0];
+ if (num_query_param_parts > 1) {
/* TODO(dgq): only the first value after the separator is considered.
* Perhaps all chars after the first separator for the query part should
* be included, even if they include the separator. */
- uri->query_parts_values[i] =
- grpc_dump_slice(query_param_parts.slices[1], GPR_DUMP_ASCII);
+ uri->query_parts_values[i] = query_param_parts[1];
} else {
uri->query_parts_values[i] = NULL;
}
- grpc_slice_buffer_reset_and_unref(&query_param_parts);
+ for (size_t j = 2; j < num_query_param_parts; j++) {
+ gpr_free(query_param_parts[j]);
+ }
+ gpr_free(query_param_parts);
+ gpr_free(full);
}
- grpc_slice_buffer_destroy(&query_parts);
- grpc_slice_buffer_destroy(&query_param_parts);
- grpc_slice_unref(query_slice);
}
grpc_uri *grpc_uri_parse(const char *uri_text, int suppress_errors) {
generated by cgit v1.2.3 (git 2.39.1) at 2024-08-07 10:17:54 +0000