aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/surface/server.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib/surface/server.c')
-rw-r--r--src/core/lib/surface/server.c100
1 files changed, 59 insertions, 41 deletions
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index 2f108af48a..7300d79b9f 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -149,6 +149,7 @@ struct call_data {
grpc_metadata_batch *recv_initial_metadata;
bool recv_idempotent_request;
+ bool recv_cacheable_request;
grpc_metadata_array initial_metadata;
request_matcher *request_matcher;
@@ -273,27 +274,24 @@ static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg,
static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
int send_goaway, grpc_error *send_disconnect) {
- grpc_transport_op op;
- struct shutdown_cleanup_args *sc;
+ struct shutdown_cleanup_args *sc = gpr_malloc(sizeof(*sc));
+ grpc_closure_init(&sc->closure, shutdown_cleanup, sc);
+ grpc_transport_op *op = grpc_make_transport_op(&sc->closure);
grpc_channel_element *elem;
- memset(&op, 0, sizeof(op));
- op.send_goaway = send_goaway;
- sc = gpr_malloc(sizeof(*sc));
+ op->send_goaway = send_goaway;
sc->slice = gpr_slice_from_copied_string("Server shutdown");
- op.goaway_message = &sc->slice;
- op.goaway_status = GRPC_STATUS_OK;
- op.disconnect_with_error = send_disconnect;
- grpc_closure_init(&sc->closure, shutdown_cleanup, sc);
- op.on_consumed = &sc->closure;
+ op->goaway_message = &sc->slice;
+ op->goaway_status = GRPC_STATUS_OK;
+ op->disconnect_with_error = send_disconnect;
elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
- elem->filter->start_transport_op(exec_ctx, elem, &op);
+ elem->filter->start_transport_op(exec_ctx, elem, op);
}
static void channel_broadcaster_shutdown(grpc_exec_ctx *exec_ctx,
channel_broadcaster *cb,
- int send_goaway,
+ bool send_goaway,
grpc_error *force_disconnect) {
size_t i;
@@ -431,7 +429,8 @@ static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd,
server_unref(exec_ctx, server);
}
-static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) {
+static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand,
+ grpc_error *error) {
if (is_channel_orphaned(chand)) return;
GPR_ASSERT(chand->server != NULL);
orphan_channel(chand);
@@ -440,14 +439,20 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) {
chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
chand->finish_destroy_channel_closure.cb_arg = chand;
- grpc_transport_op op;
- memset(&op, 0, sizeof(op));
- op.set_accept_stream = true;
- op.on_consumed = &chand->finish_destroy_channel_closure;
+ grpc_transport_op *op =
+ grpc_make_transport_op(&chand->finish_destroy_channel_closure);
+ op->set_accept_stream = true;
grpc_channel_next_op(exec_ctx,
grpc_channel_stack_element(
grpc_channel_get_channel_stack(chand->channel), 0),
- &op);
+ op);
+
+ if (error != GRPC_ERROR_NONE) {
+ const char *msg = grpc_error_string(error);
+ gpr_log(GPR_INFO, "Disconnected client: %s", msg);
+ grpc_error_free_string(msg);
+ }
+ GRPC_ERROR_UNREF(error);
}
static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) {
@@ -497,9 +502,12 @@ static void publish_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
&rc->data.batch.details->method_capacity, calld->path);
rc->data.batch.details->deadline = calld->deadline;
rc->data.batch.details->flags =
- 0 | (calld->recv_idempotent_request
- ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST
- : 0);
+ (calld->recv_idempotent_request
+ ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST
+ : 0) |
+ (calld->recv_cacheable_request
+ ? GRPC_INITIAL_METADATA_CACHEABLE_REQUEST
+ : 0);
break;
case REGISTERED_CALL:
*rc->data.registered.deadline = calld->deadline;
@@ -779,6 +787,7 @@ static void server_mutate_op(grpc_call_element *elem,
calld->on_done_recv_initial_metadata = op->recv_initial_metadata_ready;
op->recv_initial_metadata_ready = &calld->server_on_recv_initial_metadata;
op->recv_idempotent_request = &calld->recv_idempotent_request;
+ op->recv_cacheable_request = &calld->recv_cacheable_request;
}
}
@@ -840,24 +849,24 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
channel_data *chand = cd;
grpc_server *server = chand->server;
if (chand->connectivity_state != GRPC_CHANNEL_SHUTDOWN) {
- grpc_transport_op op;
- memset(&op, 0, sizeof(op));
- op.on_connectivity_state_change = &chand->channel_connectivity_changed,
- op.connectivity_state = &chand->connectivity_state;
+ grpc_transport_op *op = grpc_make_transport_op(NULL);
+ op->on_connectivity_state_change = &chand->channel_connectivity_changed,
+ op->connectivity_state = &chand->connectivity_state;
grpc_channel_next_op(exec_ctx,
grpc_channel_stack_element(
grpc_channel_get_channel_stack(chand->channel), 0),
- &op);
+ op);
} else {
gpr_mu_lock(&server->mu_global);
- destroy_channel(exec_ctx, chand);
+ destroy_channel(exec_ctx, chand, GRPC_ERROR_REF(error));
gpr_mu_unlock(&server->mu_global);
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "connectivity");
}
}
-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;
channel_data *chand = elem->channel_data;
memset(calld, 0, sizeof(call_data));
@@ -869,10 +878,12 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
server_on_recv_initial_metadata, elem);
server_ref(chand->server);
+ return GRPC_ERROR_NONE;
}
static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
- const grpc_call_stats *stats, void *ignored) {
+ const grpc_call_final_info *final_info,
+ void *ignored) {
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
@@ -1095,6 +1106,12 @@ void grpc_server_start(grpc_server *server) {
grpc_exec_ctx_finish(&exec_ctx);
}
+void grpc_server_get_pollsets(grpc_server *server, grpc_pollset ***pollsets,
+ size_t *pollset_count) {
+ *pollset_count = server->cq_count;
+ *pollsets = server->pollsets;
+}
+
void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
grpc_transport *transport,
grpc_pollset *accepting_pollset,
@@ -1111,7 +1128,7 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
size_t slots;
uint32_t probes;
uint32_t max_probes = 0;
- grpc_transport_op op;
+ grpc_transport_op *op = NULL;
channel =
grpc_channel_create(exec_ctx, NULL, args, GRPC_SERVER_CHANNEL, transport);
@@ -1171,16 +1188,16 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
gpr_mu_unlock(&s->mu_global);
GRPC_CHANNEL_INTERNAL_REF(channel, "connectivity");
- memset(&op, 0, sizeof(op));
- op.set_accept_stream = true;
- op.set_accept_stream_fn = accept_stream;
- op.set_accept_stream_user_data = chand;
- op.on_connectivity_state_change = &chand->channel_connectivity_changed;
- op.connectivity_state = &chand->connectivity_state;
+ op = grpc_make_transport_op(NULL);
+ op->set_accept_stream = true;
+ op->set_accept_stream_fn = accept_stream;
+ op->set_accept_stream_user_data = chand;
+ op->on_connectivity_state_change = &chand->channel_connectivity_changed;
+ op->connectivity_state = &chand->connectivity_state;
if (gpr_atm_acq_load(&s->shutdown_flag) != 0) {
- op.disconnect_with_error = GRPC_ERROR_CREATE("Server shutdown");
+ op->disconnect_with_error = GRPC_ERROR_CREATE("Server shutdown");
}
- grpc_transport_perform_op(exec_ctx, transport, &op);
+ grpc_transport_perform_op(exec_ctx, transport, op);
}
void done_published_shutdown(grpc_exec_ctx *exec_ctx, void *done_arg,
@@ -1249,7 +1266,8 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
l->destroy(&exec_ctx, server, l->arg, &l->destroy_done);
}
- channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 1, 0);
+ channel_broadcaster_shutdown(&exec_ctx, &broadcaster, true /* send_goaway */,
+ GRPC_ERROR_NONE);
done:
grpc_exec_ctx_finish(&exec_ctx);
@@ -1265,7 +1283,7 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
channel_broadcaster_init(server, &broadcaster);
gpr_mu_unlock(&server->mu_global);
- channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 0,
+ channel_broadcaster_shutdown(&exec_ctx, &broadcaster, false /* send_goaway */,
GRPC_ERROR_CREATE("Cancelling all calls"));
grpc_exec_ctx_finish(&exec_ctx);
}
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-18 13:30:15 +0000