aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/surface
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/surface')
-rw-r--r--src/core/surface/byte_buffer_reader.c8
-rw-r--r--src/core/surface/call.c125
-rw-r--r--src/core/surface/call.h25
-rw-r--r--src/core/surface/channel.c20
-rw-r--r--src/core/surface/channel.h13
-rw-r--r--src/core/surface/channel_create.c16
-rw-r--r--src/core/surface/completion_queue.c61
-rw-r--r--src/core/surface/completion_queue.h15
-rw-r--r--src/core/surface/lame_client.c9
-rw-r--r--src/core/surface/secure_channel_create.c18
-rw-r--r--src/core/surface/server.c172
11 files changed, 307 insertions, 175 deletions
diff --git a/src/core/surface/byte_buffer_reader.c b/src/core/surface/byte_buffer_reader.c
index 86829a686f..283db83833 100644
--- a/src/core/surface/byte_buffer_reader.c
+++ b/src/core/surface/byte_buffer_reader.c
@@ -64,11 +64,11 @@ void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
grpc_msg_decompress(reader->buffer_in->data.raw.compression,
&reader->buffer_in->data.raw.slice_buffer,
&decompressed_slices_buffer);
- reader->buffer_out = grpc_raw_byte_buffer_create(
- decompressed_slices_buffer.slices,
- decompressed_slices_buffer.count);
+ reader->buffer_out =
+ grpc_raw_byte_buffer_create(decompressed_slices_buffer.slices,
+ decompressed_slices_buffer.count);
gpr_slice_buffer_destroy(&decompressed_slices_buffer);
- } else { /* not compressed, use the input buffer as output */
+ } else { /* not compressed, use the input buffer as output */
reader->buffer_out = reader->buffer_in;
}
reader->current.index = 0;
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 5cdd7cd0f6..cf0a595147 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -99,6 +99,8 @@ typedef enum {
/* Status came from 'the wire' - or somewhere below the surface
layer */
STATUS_FROM_WIRE,
+ /* Status came from the server sending status */
+ STATUS_FROM_SERVER_STATUS,
STATUS_SOURCE_COUNT
} status_source;
@@ -152,9 +154,13 @@ struct grpc_call {
gpr_uint8 num_completed_requests;
/* are we currently reading a message? */
gpr_uint8 reading_message;
+ /* have we bound a pollset yet? */
+ gpr_uint8 bound_pollset;
/* flags with bits corresponding to write states allowing us to determine
what was sent */
gpr_uint16 last_send_contains;
+ /* cancel with this status on the next outgoing transport op */
+ grpc_status_code cancel_with_status;
/* Active ioreqs.
request_set and request_data contain one element per active ioreq
@@ -248,8 +254,10 @@ static void execute_op(grpc_call *call, grpc_transport_op *op);
static void recv_metadata(grpc_call *call, grpc_metadata_batch *metadata);
static void finish_read_ops(grpc_call *call);
static grpc_call_error cancel_with_status(grpc_call *c, grpc_status_code status,
- const char *description,
- gpr_uint8 locked);
+ const char *description);
+
+static void lock(grpc_call *call);
+static void unlock(grpc_call *call);
grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
const void *server_transport_data,
@@ -266,6 +274,9 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
gpr_mu_init(&call->mu);
call->channel = channel;
call->cq = cq;
+ if (cq) {
+ GRPC_CQ_INTERNAL_REF(cq, "bind");
+ }
call->is_client = server_transport_data == NULL;
for (i = 0; i < GRPC_IOREQ_OP_COUNT; i++) {
call->request_set[i] = REQSET_EMPTY;
@@ -282,7 +293,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
}
call->send_initial_metadata_count = add_initial_metadata_count;
call->send_deadline = send_deadline;
- grpc_channel_internal_ref(channel);
+ GRPC_CHANNEL_INTERNAL_REF(channel, "call");
call->metadata_context = grpc_channel_get_metadata_context(channel);
grpc_sopb_init(&call->send_ops);
grpc_sopb_init(&call->recv_ops);
@@ -312,7 +323,12 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
void grpc_call_set_completion_queue(grpc_call *call,
grpc_completion_queue *cq) {
+ lock(call);
call->cq = cq;
+ if (cq) {
+ GRPC_CQ_INTERNAL_REF(cq, "bind");
+ }
+ unlock(call);
}
grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) {
@@ -333,7 +349,7 @@ static void destroy_call(void *call, int ignored_success) {
size_t i;
grpc_call *c = call;
grpc_call_stack_destroy(CALL_STACK_FROM_CALL(c));
- grpc_channel_internal_unref(c->channel);
+ GRPC_CHANNEL_INTERNAL_UNREF(c->channel, "call");
gpr_mu_destroy(&c->mu);
for (i = 0; i < STATUS_SOURCE_COUNT; i++) {
if (c->status[i].details) {
@@ -359,6 +375,9 @@ static void destroy_call(void *call, int ignored_success) {
grpc_sopb_destroy(&c->recv_ops);
grpc_bbq_destroy(&c->incoming_queue);
gpr_slice_buffer_destroy(&c->incoming_message);
+ if (c->cq) {
+ GRPC_CQ_INTERNAL_UNREF(c->cq, "bind");
+ }
gpr_free(c);
}
@@ -411,6 +430,7 @@ static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); }
static int need_more_data(grpc_call *call) {
if (call->read_state == READ_STATE_STREAM_CLOSED) return 0;
+ /* TODO(ctiller): this needs some serious cleanup */
return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) ||
(is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) &&
grpc_bbq_empty(&call->incoming_queue)) ||
@@ -419,7 +439,8 @@ static int need_more_data(grpc_call *call) {
is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS) ||
(is_op_live(call, GRPC_IOREQ_RECV_CLOSE) &&
grpc_bbq_empty(&call->incoming_queue)) ||
- (call->write_state == WRITE_STATE_INITIAL && !call->is_client);
+ (call->write_state == WRITE_STATE_INITIAL && !call->is_client) ||
+ (call->cancel_with_status != GRPC_STATUS_OK);
}
static void unlock(grpc_call *call) {
@@ -431,6 +452,10 @@ static void unlock(grpc_call *call) {
memset(&op, 0, sizeof(op));
+ op.cancel_with_status = call->cancel_with_status;
+ start_op = op.cancel_with_status != GRPC_STATUS_OK;
+ call->cancel_with_status = GRPC_STATUS_OK; /* reset */
+
if (!call->receiving && need_more_data(call)) {
op.recv_ops = &call->recv_ops;
op.recv_state = &call->recv_state;
@@ -449,6 +474,12 @@ static void unlock(grpc_call *call) {
}
}
+ if (!call->bound_pollset && call->cq && (!call->is_client || start_op)) {
+ call->bound_pollset = 1;
+ op.bind_pollset = grpc_cq_pollset(call->cq);
+ start_op = 1;
+ }
+
if (!call->completing && call->num_completed_requests != 0) {
completing_requests = call->num_completed_requests;
memcpy(completed_requests, call->completed_requests,
@@ -551,10 +582,18 @@ static void finish_live_ioreq_op(grpc_call *call, grpc_ioreq_op op,
call->write_state = WRITE_STATE_WRITE_CLOSED;
}
break;
+ case GRPC_IOREQ_SEND_STATUS:
+ if (call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details !=
+ NULL) {
+ grpc_mdstr_unref(
+ call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details);
+ call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details =
+ NULL;
+ }
+ break;
case GRPC_IOREQ_RECV_CLOSE:
case GRPC_IOREQ_SEND_INITIAL_METADATA:
case GRPC_IOREQ_SEND_TRAILING_METADATA:
- case GRPC_IOREQ_SEND_STATUS:
case GRPC_IOREQ_SEND_CLOSE:
break;
case GRPC_IOREQ_RECV_STATUS:
@@ -655,7 +694,7 @@ static int begin_message(grpc_call *call, grpc_begin_message msg) {
gpr_asprintf(
&message, "Message terminated early; read %d bytes, expected %d",
(int)call->incoming_message.length, (int)call->incoming_message_length);
- cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
+ cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
gpr_free(message);
return 0;
}
@@ -666,7 +705,7 @@ static int begin_message(grpc_call *call, grpc_begin_message msg) {
&message,
"Maximum message length of %d exceeded by a message of length %d",
grpc_channel_get_max_message_length(call->channel), msg.length);
- cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
+ cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
gpr_free(message);
return 0;
} else if (msg.length > 0) {
@@ -688,7 +727,7 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) {
/* we have to be reading a message to know what to do here */
if (!call->reading_message) {
cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT,
- "Received payload data while not reading a message", 1);
+ "Received payload data while not reading a message");
return 0;
}
/* append the slice to the incoming buffer */
@@ -699,7 +738,7 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) {
gpr_asprintf(
&message, "Receiving message overflow; read %d bytes, expected %d",
(int)call->incoming_message.length, (int)call->incoming_message_length);
- cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message, 1);
+ cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message);
gpr_free(message);
return 0;
} else if (call->incoming_message.length == call->incoming_message_length) {
@@ -841,7 +880,6 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) {
}
grpc_sopb_add_metadata(&call->send_ops, mdb);
op->send_ops = &call->send_ops;
- op->bind_pollset = grpc_cq_pollset(call->cq);
call->last_send_contains |= 1 << GRPC_IOREQ_SEND_INITIAL_METADATA;
call->send_initial_metadata_count = 0;
/* fall through intended */
@@ -880,8 +918,9 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) {
call->metadata_context,
grpc_mdstr_ref(
grpc_channel_get_message_string(call->channel)),
- grpc_mdstr_from_string(call->metadata_context,
- data.send_status.details)));
+ data.send_status.details));
+ call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details =
+ NULL;
}
grpc_sopb_add_metadata(&call->send_ops, mdb);
}
@@ -981,6 +1020,14 @@ static grpc_call_error start_ioreq(grpc_call *call, const grpc_ioreq *reqs,
GRPC_CALL_ERROR_INVALID_METADATA);
}
}
+ if (op == GRPC_IOREQ_SEND_STATUS) {
+ set_status_code(call, STATUS_FROM_SERVER_STATUS,
+ reqs[i].data.send_status.code);
+ if (reqs[i].data.send_status.details) {
+ set_status_details(call, STATUS_FROM_SERVER_STATUS,
+ grpc_mdstr_ref(reqs[i].data.send_status.details));
+ }
+ }
have_ops |= 1u << op;
call->request_data[op] = data;
@@ -1031,35 +1078,43 @@ grpc_call_error grpc_call_cancel(grpc_call *call) {
grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
grpc_status_code status,
const char *description) {
- return cancel_with_status(c, status, description, 0);
+ grpc_call_error r;
+ lock(c);
+ r = cancel_with_status(c, status, description);
+ unlock(c);
+ return r;
}
static grpc_call_error cancel_with_status(grpc_call *c, grpc_status_code status,
- const char *description,
- gpr_uint8 locked) {
- grpc_transport_op op;
+ const char *description) {
grpc_mdstr *details =
description ? grpc_mdstr_from_string(c->metadata_context, description)
: NULL;
- memset(&op, 0, sizeof(op));
- op.cancel_with_status = status;
- if (locked == 0) {
- lock(c);
- }
+ GPR_ASSERT(status != GRPC_STATUS_OK);
+
set_status_code(c, STATUS_FROM_API_OVERRIDE, status);
set_status_details(c, STATUS_FROM_API_OVERRIDE, details);
- if (locked == 0) {
- unlock(c);
- }
- execute_op(c, &op);
+ c->cancel_with_status = status;
return GRPC_CALL_OK;
}
+static void finished_loose_op(void *call, int success_ignored) {
+ GRPC_CALL_INTERNAL_UNREF(call, "loose-op", 0);
+}
+
static void execute_op(grpc_call *call, grpc_transport_op *op) {
grpc_call_element *elem;
+
+ GPR_ASSERT(op->on_consumed == NULL);
+ if (op->cancel_with_status != GRPC_STATUS_OK || op->bind_pollset) {
+ GRPC_CALL_INTERNAL_REF(call, "loose-op");
+ op->on_consumed = finished_loose_op;
+ op->on_consumed_user_data = call;
+ }
+
elem = CALL_ELEM_FROM_CALL(call, 0);
op->context = call->context;
elem->filter->start_transport_op(elem, op);
@@ -1072,12 +1127,10 @@ grpc_call *grpc_call_from_top_element(grpc_call_element *elem) {
static void call_alarm(void *arg, int success) {
grpc_call *call = arg;
if (success) {
- if (call->is_client) {
- cancel_with_status(call, GRPC_STATUS_DEADLINE_EXCEEDED,
- "Deadline Exceeded", 0);
- } else {
- grpc_call_cancel(call);
- }
+ lock(call);
+ cancel_with_status(call, GRPC_STATUS_DEADLINE_EXCEEDED,
+ "Deadline Exceeded");
+ unlock(call);
}
GRPC_CALL_INTERNAL_UNREF(call, "alarm", 1);
}
@@ -1235,7 +1288,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
req->flags = op->flags;
break;
case GRPC_OP_SEND_MESSAGE:
- if (!are_write_flags_valid(op->flags)){
+ if (!are_write_flags_valid(op->flags)) {
return GRPC_CALL_ERROR_INVALID_FLAGS;
}
req = &reqs[out++];
@@ -1270,7 +1323,11 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
req->op = GRPC_IOREQ_SEND_STATUS;
req->data.send_status.code = op->data.send_status_from_server.status;
req->data.send_status.details =
- op->data.send_status_from_server.status_details;
+ op->data.send_status_from_server.status_details != NULL
+ ? grpc_mdstr_from_string(
+ call->metadata_context,
+ op->data.send_status_from_server.status_details)
+ : NULL;
req = &reqs[out++];
req->op = GRPC_IOREQ_SEND_CLOSE;
break;
diff --git a/src/core/surface/call.h b/src/core/surface/call.h
index 7a14161253..fb3662b50d 100644
--- a/src/core/surface/call.h
+++ b/src/core/surface/call.h
@@ -72,14 +72,14 @@ typedef union {
grpc_byte_buffer *send_message;
struct {
grpc_status_code code;
- const char *details;
+ grpc_mdstr *details;
} send_status;
} grpc_ioreq_data;
typedef struct {
grpc_ioreq_op op;
grpc_ioreq_data data;
- gpr_uint32 flags; /**< A copy of the write flags from grpc_op */
+ gpr_uint32 flags; /**< A copy of the write flags from grpc_op */
} grpc_ioreq;
typedef void (*grpc_ioreq_completion_func)(grpc_call *call, int success,
@@ -96,8 +96,10 @@ grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call);
#ifdef GRPC_CALL_REF_COUNT_DEBUG
void grpc_call_internal_ref(grpc_call *call, const char *reason);
-void grpc_call_internal_unref(grpc_call *call, const char *reason, int allow_immediate_deletion);
-#define GRPC_CALL_INTERNAL_REF(call, reason) grpc_call_internal_ref(call, reason)
+void grpc_call_internal_unref(grpc_call *call, const char *reason,
+ int allow_immediate_deletion);
+#define GRPC_CALL_INTERNAL_REF(call, reason) \
+ grpc_call_internal_ref(call, reason)
#define GRPC_CALL_INTERNAL_UNREF(call, reason, allow_immediate_deletion) \
grpc_call_internal_unref(call, reason, allow_immediate_deletion)
#else
@@ -125,8 +127,7 @@ void grpc_call_log_batch(char *file, int line, gpr_log_severity severity,
void grpc_server_log_request_call(char *file, int line,
gpr_log_severity severity,
- grpc_server *server,
- grpc_call **call,
+ grpc_server *server, grpc_call **call,
grpc_call_details *details,
grpc_metadata_array *initial_metadata,
grpc_completion_queue *cq_bound_to_call,
@@ -135,16 +136,20 @@ void grpc_server_log_request_call(char *file, int line,
/* Set a context pointer.
No thread safety guarantees are made wrt this value. */
-void grpc_call_context_set(grpc_call *call, grpc_context_index elem, void *value,
- void (*destroy)(void *value));
+void grpc_call_context_set(grpc_call *call, grpc_context_index elem,
+ void *value, void (*destroy)(void *value));
/* Get a context pointer. */
void *grpc_call_context_get(grpc_call *call, grpc_context_index elem);
#define GRPC_CALL_LOG_BATCH(sev, call, ops, nops, tag) \
if (grpc_trace_batch) grpc_call_log_batch(sev, call, ops, nops, tag)
-#define GRPC_SERVER_LOG_REQUEST_CALL(sev, server, call, details, initial_metadata, cq_bound_to_call, cq_for_notifications, tag) \
- if (grpc_trace_batch) grpc_server_log_request_call(sev, server, call, details, initial_metadata, cq_bound_to_call, cq_for_notifications, tag)
+#define GRPC_SERVER_LOG_REQUEST_CALL(sev, server, call, details, \
+ initial_metadata, cq_bound_to_call, \
+ cq_for_notifications, tag) \
+ if (grpc_trace_batch) \
+ grpc_server_log_request_call(sev, server, call, details, initial_metadata, \
+ cq_bound_to_call, cq_for_notifications, tag)
gpr_uint8 grpc_call_is_client(grpc_call *call);
diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c
index 9175ad0572..9ecdd374a9 100644
--- a/src/core/surface/channel.c
+++ b/src/core/surface/channel.c
@@ -187,8 +187,14 @@ grpc_call *grpc_channel_create_registered_call(
grpc_mdelem_ref(rc->authority), deadline);
}
-void grpc_channel_internal_ref(grpc_channel *channel) {
- gpr_ref(&channel->refs);
+#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG
+void grpc_channel_internal_ref(grpc_channel *c, const char *reason) {
+ gpr_log(GPR_DEBUG, "CHANNEL: ref %p %d -> %d [%s]", c, c->refs.count,
+ c->refs.count + 1, reason);
+#else
+void grpc_channel_internal_ref(grpc_channel *c) {
+#endif
+ gpr_ref(&c->refs);
}
static void destroy_channel(void *p, int ok) {
@@ -214,7 +220,13 @@ static void destroy_channel(void *p, int ok) {
gpr_free(channel);
}
+#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG
+void grpc_channel_internal_unref(grpc_channel *channel, const char *reason) {
+ gpr_log(GPR_DEBUG, "CHANNEL: unref %p %d -> %d [%s]", channel,
+ channel->refs.count, channel->refs.count - 1, reason);
+#else
void grpc_channel_internal_unref(grpc_channel *channel) {
+#endif
if (gpr_unref(&channel->refs)) {
channel->destroy_closure.cb = destroy_channel;
channel->destroy_closure.cb_arg = channel;
@@ -238,11 +250,11 @@ void grpc_channel_destroy(grpc_channel *channel) {
op.dir = GRPC_CALL_DOWN;
elem->filter->channel_op(elem, NULL, &op);
- grpc_channel_internal_unref(channel);
+ GRPC_CHANNEL_INTERNAL_UNREF(channel, "channel");
}
void grpc_client_channel_closed(grpc_channel_element *elem) {
- grpc_channel_internal_unref(CHANNEL_FROM_TOP_ELEM(elem));
+ GRPC_CHANNEL_INTERNAL_UNREF(CHANNEL_FROM_TOP_ELEM(elem), "closed");
}
grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel) {
diff --git a/src/core/surface/channel.h b/src/core/surface/channel.h
index 6d1ed87900..ba3c0abac9 100644
--- a/src/core/surface/channel.h
+++ b/src/core/surface/channel.h
@@ -58,7 +58,20 @@ gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel);
void grpc_client_channel_closed(grpc_channel_element *elem);
+#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG
+void grpc_channel_internal_ref(grpc_channel *channel, const char *reason);
+void grpc_channel_internal_unref(grpc_channel *channel, const char *reason);
+#define GRPC_CHANNEL_INTERNAL_REF(channel, reason) \
+ grpc_channel_internal_ref(channel, reason)
+#define GRPC_CHANNEL_INTERNAL_UNREF(channel, reason) \
+ grpc_channel_internal_unref(channel, reason)
+#else
void grpc_channel_internal_ref(grpc_channel *channel);
void grpc_channel_internal_unref(grpc_channel *channel);
+#define GRPC_CHANNEL_INTERNAL_REF(channel, reason) \
+ grpc_channel_internal_ref(channel)
+#define GRPC_CHANNEL_INTERNAL_UNREF(channel, reason) \
+ grpc_channel_internal_unref(channel)
+#endif
#endif /* GRPC_INTERNAL_CORE_SURFACE_CHANNEL_H */
diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c
index 946ee0949d..d069a04a9a 100644
--- a/src/core/surface/channel_create.c
+++ b/src/core/surface/channel_create.c
@@ -91,7 +91,7 @@ static void done(request *r, int was_successful) {
static void on_connect(void *rp, grpc_endpoint *tcp) {
request *r = rp;
- if (!grpc_client_setup_request_should_continue(r->cs_request)) {
+ if (!grpc_client_setup_request_should_continue(r->cs_request, "on_connect")) {
if (tcp) {
grpc_endpoint_shutdown(tcp);
grpc_endpoint_destroy(tcp);
@@ -107,12 +107,12 @@ static void on_connect(void *rp, grpc_endpoint *tcp) {
} else {
return;
}
- } else if (grpc_client_setup_cb_begin(r->cs_request)) {
+ } else if (grpc_client_setup_cb_begin(r->cs_request, "on_connect")) {
grpc_create_chttp2_transport(
r->setup->setup_callback, r->setup->setup_user_data,
grpc_client_setup_get_channel_args(r->cs_request), tcp, NULL, 0,
grpc_client_setup_get_mdctx(r->cs_request), 1);
- grpc_client_setup_cb_end(r->cs_request);
+ grpc_client_setup_cb_end(r->cs_request, "on_connect");
done(r, 1);
return;
} else {
@@ -126,9 +126,10 @@ static int maybe_try_next_resolved(request *r) {
if (!r->resolved) return 0;
if (r->resolved_index == r->resolved->naddrs) return 0;
addr = &r->resolved->addrs[r->resolved_index++];
- grpc_tcp_client_connect(on_connect, r, (struct sockaddr *)&addr->addr,
- addr->len,
- grpc_client_setup_request_deadline(r->cs_request));
+ grpc_tcp_client_connect(
+ on_connect, r, grpc_client_setup_get_interested_parties(r->cs_request),
+ (struct sockaddr *)&addr->addr, addr->len,
+ grpc_client_setup_request_deadline(r->cs_request));
return 1;
}
@@ -137,7 +138,8 @@ static void on_resolved(void *rp, grpc_resolved_addresses *resolved) {
request *r = rp;
/* if we're not still the active request, abort */
- if (!grpc_client_setup_request_should_continue(r->cs_request)) {
+ if (!grpc_client_setup_request_should_continue(r->cs_request,
+ "on_resolved")) {
if (resolved) {
grpc_resolved_addresses_destroy(resolved);
}
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index b48fbace31..bd0fabf9da 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -59,9 +59,6 @@ typedef struct event {
/* Completion queue structure */
struct grpc_completion_queue {
- /* TODO(ctiller): see if this can be removed */
- int allow_polling;
-
/* When refs drops to zero, we are in shutdown mode, and will be destroyable
once all queued events are drained */
gpr_refcount refs;
@@ -76,6 +73,7 @@ struct grpc_completion_queue {
event *queue;
/* Fixed size chained hash table of events for pluck() */
event *buckets[NUM_TAG_BUCKETS];
+ int is_server_cq;
};
grpc_completion_queue *grpc_completion_queue_create(void) {
@@ -86,20 +84,31 @@ grpc_completion_queue *grpc_completion_queue_create(void) {
/* One for destroy(), one for pollset_shutdown */
gpr_ref_init(&cc->owning_refs, 2);
grpc_pollset_init(&cc->pollset);
- cc->allow_polling = 1;
return cc;
}
+#ifdef GRPC_CQ_REF_COUNT_DEBUG
+void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason) {
+ gpr_log(GPR_DEBUG, "CQ:%p ref %d -> %d %s", cc, (int)cc->owning_refs.count,
+ (int)cc->owning_refs.count + 1, reason);
+#else
void grpc_cq_internal_ref(grpc_completion_queue *cc) {
+#endif
gpr_ref(&cc->owning_refs);
}
static void on_pollset_destroy_done(void *arg) {
grpc_completion_queue *cc = arg;
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy");
}
+#ifdef GRPC_CQ_REF_COUNT_DEBUG
+void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason) {
+ gpr_log(GPR_DEBUG, "CQ:%p unref %d -> %d %s", cc, (int)cc->owning_refs.count,
+ (int)cc->owning_refs.count - 1, reason);
+#else
void grpc_cq_internal_unref(grpc_completion_queue *cc) {
+#endif
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->queue == NULL);
grpc_pollset_destroy(&cc->pollset);
@@ -107,10 +116,6 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) {
}
}
-void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc) {
- cc->allow_polling = 0;
-}
-
/* Create and append an event to the queue. Returns the event so that its data
members can be filled in.
Requires GRPC_POLLSET_MU(&cc->pollset) locked. */
@@ -134,7 +139,6 @@ static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type,
ev->bucket_prev = cc->buckets[bucket]->bucket_prev;
ev->bucket_next->bucket_prev = ev->bucket_prev->bucket_next = ev;
}
- gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
grpc_pollset_kick(&cc->pollset);
return ev;
}
@@ -157,7 +161,6 @@ void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
GPR_ASSERT(!cc->shutdown);
GPR_ASSERT(cc->shutdown_called);
cc->shutdown = 1;
- gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
shutdown = 1;
}
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
@@ -180,7 +183,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
event *ev = NULL;
grpc_event ret;
- grpc_cq_internal_ref(cc);
+ GRPC_CQ_INTERNAL_REF(cc, "next");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
if (cc->queue != NULL) {
@@ -207,16 +210,12 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
ev = create_shutdown_event();
break;
}
- if (cc->allow_polling && grpc_pollset_work(&cc->pollset, deadline)) {
- continue;
- }
- if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset),
- GRPC_POLLSET_MU(&cc->pollset), deadline)) {
+ if (!grpc_pollset_work(&cc->pollset, deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "next");
return ret;
}
}
@@ -224,7 +223,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
ret = ev->base;
gpr_free(ev);
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "next");
return ret;
}
@@ -262,7 +261,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
event *ev = NULL;
grpc_event ret;
- grpc_cq_internal_ref(cc);
+ GRPC_CQ_INTERNAL_REF(cc, "pluck");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
if ((ev = pluck_event(cc, tag))) {
@@ -272,16 +271,12 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
ev = create_shutdown_event();
break;
}
- if (cc->allow_polling && grpc_pollset_work(&cc->pollset, deadline)) {
- continue;
- }
- if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset),
- GRPC_POLLSET_MU(&cc->pollset), deadline)) {
+ if (!grpc_pollset_work(&cc->pollset, deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
return ret;
}
}
@@ -289,7 +284,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
ret = ev->base;
gpr_free(ev);
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
return ret;
}
@@ -297,6 +292,10 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
to zero here, then enter shutdown mode and wake up any waiters */
void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
+ if (cc->shutdown_called) {
+ gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+ return;
+ }
cc->shutdown_called = 1;
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
@@ -304,14 +303,14 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
GPR_ASSERT(!cc->shutdown);
cc->shutdown = 1;
- gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
}
}
void grpc_completion_queue_destroy(grpc_completion_queue *cc) {
- grpc_cq_internal_unref(cc);
+ grpc_completion_queue_shutdown(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "destroy");
}
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
@@ -325,3 +324,7 @@ void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc) {
gpr_time_add(gpr_now(), gpr_time_from_millis(100)));
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
}
+
+void grpc_cq_mark_server_cq(grpc_completion_queue *cc) { cc->is_server_cq = 1; }
+
+int grpc_cq_is_server_cq(grpc_completion_queue *cc) { return cc->is_server_cq; }
diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h
index 7b6fad98fd..e76910c00b 100644
--- a/src/core/surface/completion_queue.h
+++ b/src/core/surface/completion_queue.h
@@ -39,8 +39,17 @@
#include "src/core/iomgr/pollset.h"
#include <grpc/grpc.h>
+#ifdef GRPC_CQ_REF_COUNT_DEBUG
+void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason);
+void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason);
+#define GRPC_CQ_INTERNAL_REF(cc, reason) grpc_cq_internal_ref(cc, reason)
+#define GRPC_CQ_INTERNAL_UNREF(cc, reason) grpc_cq_internal_unref(cc, reason)
+#else
void grpc_cq_internal_ref(grpc_completion_queue *cc);
void grpc_cq_internal_unref(grpc_completion_queue *cc);
+#define GRPC_CQ_INTERNAL_REF(cc, reason) grpc_cq_internal_ref(cc)
+#define GRPC_CQ_INTERNAL_UNREF(cc, reason) grpc_cq_internal_unref(cc)
+#endif
/* Flag that an operation is beginning: the completion channel will not finish
shutdown until a corrensponding grpc_cq_end_* call is made */
@@ -50,11 +59,11 @@ void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call);
void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
int success);
-/* disable polling for some tests */
-void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc);
-
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);
void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc);
+void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
+int grpc_cq_is_server_cq(grpc_completion_queue *cc);
+
#endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */
diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c
index a3b0b2672b..b667128aef 100644
--- a/src/core/surface/lame_client.c
+++ b/src/core/surface/lame_client.c
@@ -77,6 +77,9 @@ static void lame_start_transport_op(grpc_call_element *elem,
*op->recv_state = GRPC_STREAM_CLOSED;
op->on_done_recv(op->recv_user_data, 1);
}
+ if (op->on_consumed) {
+ op->on_consumed(op->on_consumed_user_data, 0);
+ }
}
static void channel_op(grpc_channel_element *elem,
@@ -115,9 +118,9 @@ static void init_channel_elem(grpc_channel_element *elem,
static void destroy_channel_elem(grpc_channel_element *elem) {}
static const grpc_channel_filter lame_filter = {
- lame_start_transport_op, channel_op, sizeof(call_data), init_call_elem,
- destroy_call_elem, sizeof(channel_data), init_channel_elem,
- destroy_channel_elem, "lame-client",
+ lame_start_transport_op, channel_op, sizeof(call_data),
+ init_call_elem, destroy_call_elem, sizeof(channel_data),
+ init_channel_elem, destroy_channel_elem, "lame-client",
};
grpc_channel *grpc_lame_client_channel_create(void) {
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c
index 8b39934881..fae3e4e90a 100644
--- a/src/core/surface/secure_channel_create.c
+++ b/src/core/surface/secure_channel_create.c
@@ -97,12 +97,13 @@ static void on_secure_transport_setup_done(void *rp,
if (status != GRPC_SECURITY_OK) {
gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status);
done(r, 0);
- } else if (grpc_client_setup_cb_begin(r->cs_request)) {
+ } else if (grpc_client_setup_cb_begin(r->cs_request,
+ "on_secure_transport_setup_done")) {
grpc_create_chttp2_transport(
r->setup->setup_callback, r->setup->setup_user_data,
grpc_client_setup_get_channel_args(r->cs_request), secure_endpoint,
NULL, 0, grpc_client_setup_get_mdctx(r->cs_request), 1);
- grpc_client_setup_cb_end(r->cs_request);
+ grpc_client_setup_cb_end(r->cs_request, "on_secure_transport_setup_done");
done(r, 1);
} else {
done(r, 0);
@@ -113,7 +114,8 @@ static void on_secure_transport_setup_done(void *rp,
static void on_connect(void *rp, grpc_endpoint *tcp) {
request *r = rp;
- if (!grpc_client_setup_request_should_continue(r->cs_request)) {
+ if (!grpc_client_setup_request_should_continue(r->cs_request,
+ "on_connect.secure")) {
if (tcp) {
grpc_endpoint_shutdown(tcp);
grpc_endpoint_destroy(tcp);
@@ -141,9 +143,10 @@ static int maybe_try_next_resolved(request *r) {
if (!r->resolved) return 0;
if (r->resolved_index == r->resolved->naddrs) return 0;
addr = &r->resolved->addrs[r->resolved_index++];
- grpc_tcp_client_connect(on_connect, r, (struct sockaddr *)&addr->addr,
- addr->len,
- grpc_client_setup_request_deadline(r->cs_request));
+ grpc_tcp_client_connect(
+ on_connect, r, grpc_client_setup_get_interested_parties(r->cs_request),
+ (struct sockaddr *)&addr->addr, addr->len,
+ grpc_client_setup_request_deadline(r->cs_request));
return 1;
}
@@ -152,7 +155,8 @@ static void on_resolved(void *rp, grpc_resolved_addresses *resolved) {
request *r = rp;
/* if we're not still the active request, abort */
- if (!grpc_client_setup_request_should_continue(r->cs_request)) {
+ if (!grpc_client_setup_request_should_continue(r->cs_request,
+ "on_resolved.secure")) {
if (resolved) {
grpc_resolved_addresses_destroy(resolved);
}
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 825ef66804..c8ac559a0d 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -114,6 +114,7 @@ typedef struct channel_registered_method {
struct channel_data {
grpc_server *server;
+ size_t num_calls;
grpc_channel *channel;
grpc_mdstr *path_key;
grpc_mdstr *authority_key;
@@ -123,7 +124,6 @@ struct channel_data {
channel_registered_method *registered_methods;
gpr_uint32 registered_method_slots;
gpr_uint32 registered_method_max_probes;
- grpc_iomgr_closure finish_shutdown_channel_closure;
grpc_iomgr_closure finish_destroy_channel_closure;
};
@@ -198,6 +198,9 @@ struct call_data {
static void begin_call(grpc_server *server, call_data *calld,
requested_call *rc);
static void fail_call(grpc_server *server, requested_call *rc);
+static void shutdown_channel(channel_data *chand, int send_goaway,
+ int send_disconnect);
+static void maybe_finish_shutdown(grpc_server *server);
static int call_list_join(call_data **root, call_data *call, call_list list) {
GPR_ASSERT(!call->root[list]);
@@ -281,7 +284,7 @@ static void server_delete(grpc_server *server) {
gpr_free(rm);
}
for (i = 0; i < server->cq_count; i++) {
- grpc_cq_internal_unref(server->cqs[i]);
+ GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server");
}
gpr_free(server->cqs);
gpr_free(server->pollsets);
@@ -308,7 +311,7 @@ static void orphan_channel(channel_data *chand) {
static void finish_destroy_channel(void *cd, int success) {
channel_data *chand = cd;
grpc_server *server = chand->server;
- grpc_channel_internal_unref(chand->channel);
+ GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server");
server_unref(server);
}
@@ -317,6 +320,7 @@ static void destroy_channel(channel_data *chand) {
GPR_ASSERT(chand->server != NULL);
orphan_channel(chand);
server_ref(chand->server);
+ maybe_finish_shutdown(chand->server);
chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
chand->finish_destroy_channel_closure.cb_arg = chand;
grpc_iomgr_add_callback(&chand->finish_destroy_channel_closure);
@@ -397,12 +401,28 @@ static int num_listeners(grpc_server *server) {
static void maybe_finish_shutdown(grpc_server *server) {
size_t i;
- if (server->shutdown && !server->shutdown_published && server->lists[ALL_CALLS] == NULL && server->listeners_destroyed == num_listeners(server)) {
- server->shutdown_published = 1;
- for (i = 0; i < server->num_shutdown_tags; i++) {
- grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag,
- NULL, 1);
- }
+ if (!server->shutdown || server->shutdown_published) {
+ return;
+ }
+ if (server->lists[ALL_CALLS] != NULL) {
+ gpr_log(GPR_DEBUG,
+ "Waiting for all calls to finish before destroying server");
+ return;
+ }
+ if (server->root_channel_data.next != &server->root_channel_data) {
+ gpr_log(GPR_DEBUG,
+ "Waiting for all channels to close before destroying server");
+ return;
+ }
+ if (server->listeners_destroyed < num_listeners(server)) {
+ gpr_log(GPR_DEBUG, "Waiting for all listeners to be destroyed (@ %d/%d)",
+ server->listeners_destroyed, num_listeners(server));
+ return;
+ }
+ server->shutdown_published = 1;
+ for (i = 0; i < server->num_shutdown_tags; i++) {
+ grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag,
+ NULL, 1);
}
}
@@ -420,6 +440,14 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
return md;
}
+static void decrement_call_count(channel_data *chand) {
+ chand->num_calls--;
+ if (0 == chand->num_calls && chand->server->shutdown) {
+ shutdown_channel(chand, 0, 1);
+ }
+ maybe_finish_shutdown(chand->server);
+}
+
static void server_on_recv(void *ptr, int success) {
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
@@ -467,10 +495,9 @@ static void server_on_recv(void *ptr, int success) {
calld->state = ZOMBIED;
grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
grpc_iomgr_add_callback(&calld->kill_zombie_closure);
-
}
if (call_list_remove(calld, ALL_CALLS)) {
- maybe_finish_shutdown(chand->server);
+ decrement_call_count(chand);
}
gpr_mu_unlock(&chand->server->mu);
break;
@@ -531,22 +558,49 @@ static void channel_op(grpc_channel_element *elem,
}
}
-static void finish_shutdown_channel(void *cd, int success) {
- channel_data *chand = cd;
+typedef struct {
+ channel_data *chand;
+ int send_goaway;
+ int send_disconnect;
+ grpc_iomgr_closure finish_shutdown_channel_closure;
+} shutdown_channel_args;
+
+static void finish_shutdown_channel(void *p, int success) {
+ shutdown_channel_args *sca = p;
grpc_channel_op op;
- op.type = GRPC_CHANNEL_DISCONNECT;
- op.dir = GRPC_CALL_DOWN;
- channel_op(grpc_channel_stack_element(
- grpc_channel_get_channel_stack(chand->channel), 0),
- NULL, &op);
- grpc_channel_internal_unref(chand->channel);
+
+ if (sca->send_goaway) {
+ op.type = GRPC_CHANNEL_GOAWAY;
+ op.dir = GRPC_CALL_DOWN;
+ op.data.goaway.status = GRPC_STATUS_OK;
+ op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown");
+ channel_op(grpc_channel_stack_element(
+ grpc_channel_get_channel_stack(sca->chand->channel), 0),
+ NULL, &op);
+ }
+ if (sca->send_disconnect) {
+ op.type = GRPC_CHANNEL_DISCONNECT;
+ op.dir = GRPC_CALL_DOWN;
+ channel_op(grpc_channel_stack_element(
+ grpc_channel_get_channel_stack(sca->chand->channel), 0),
+ NULL, &op);
+ }
+ GRPC_CHANNEL_INTERNAL_UNREF(sca->chand->channel, "shutdown");
+
+ gpr_free(sca);
}
-static void shutdown_channel(channel_data *chand) {
- grpc_channel_internal_ref(chand->channel);
- chand->finish_shutdown_channel_closure.cb = finish_shutdown_channel;
- chand->finish_shutdown_channel_closure.cb_arg = chand;
- grpc_iomgr_add_callback(&chand->finish_shutdown_channel_closure);
+static void shutdown_channel(channel_data *chand, int send_goaway,
+ int send_disconnect) {
+ shutdown_channel_args *sca;
+ GRPC_CHANNEL_INTERNAL_REF(chand->channel, "shutdown");
+ sca = gpr_malloc(sizeof(shutdown_channel_args));
+ sca->chand = chand;
+ sca->send_goaway = send_goaway;
+ sca->send_disconnect = send_disconnect;
+ sca->finish_shutdown_channel_closure.cb = finish_shutdown_channel;
+ sca->finish_shutdown_channel_closure.cb_arg = sca;
+ grpc_iomgr_add_callback(&sca->finish_shutdown_channel_closure);
}
static void init_call_elem(grpc_call_element *elem,
@@ -560,6 +614,7 @@ static void init_call_elem(grpc_call_element *elem,
gpr_mu_lock(&chand->server->mu);
call_list_join(&chand->server->lists[ALL_CALLS], calld, ALL_CALLS);
+ chand->num_calls++;
gpr_mu_unlock(&chand->server->mu);
server_ref(chand->server);
@@ -578,7 +633,7 @@ static void destroy_call_elem(grpc_call_element *elem) {
removed[i] = call_list_remove(elem->call_data, i);
}
if (removed[ALL_CALLS]) {
- maybe_finish_shutdown(chand->server);
+ decrement_call_count(chand);
}
gpr_mu_unlock(&chand->server->mu);
@@ -600,6 +655,7 @@ static void init_channel_elem(grpc_channel_element *elem,
GPR_ASSERT(is_first);
GPR_ASSERT(!is_last);
chand->server = NULL;
+ chand->num_calls = 0;
chand->channel = NULL;
chand->path_key = grpc_mdstr_from_string(metadata_context, ":path");
chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority");
@@ -626,6 +682,7 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
chand->next->prev = chand->prev;
chand->prev->next = chand->next;
chand->next = chand->prev = chand;
+ maybe_finish_shutdown(chand->server);
gpr_mu_unlock(&chand->server->mu);
grpc_mdstr_unref(chand->path_key);
grpc_mdstr_unref(chand->authority_key);
@@ -651,7 +708,8 @@ void grpc_server_register_completion_queue(grpc_server *server,
for (i = 0; i < server->cq_count; i++) {
if (server->cqs[i] == cq) return;
}
- grpc_cq_internal_ref(cq);
+ GRPC_CQ_INTERNAL_REF(cq, "server");
+ grpc_cq_mark_server_cq(cq);
n = server->cq_count++;
server->cqs = gpr_realloc(server->cqs,
server->cq_count * sizeof(grpc_completion_queue *));
@@ -713,7 +771,8 @@ void *grpc_server_register_method(grpc_server *server, const char *method,
const char *host) {
registered_method *m;
if (!method) {
- gpr_log(GPR_ERROR, "grpc_server_register_method method string cannot be NULL");
+ gpr_log(GPR_ERROR,
+ "grpc_server_register_method method string cannot be NULL");
return NULL;
}
for (m = server->registered_methods; m; m = m->next) {
@@ -836,12 +895,8 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
grpc_completion_queue *cq, void *tag) {
listener *l;
requested_call_array requested_calls;
- channel_data **channels;
channel_data *c;
- size_t nchannels;
size_t i;
- grpc_channel_op op;
- grpc_channel_element *elem;
registered_method *rm;
shutdown_tag *sdt;
@@ -859,18 +914,9 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
return;
}
- nchannels = 0;
for (c = server->root_channel_data.next; c != &server->root_channel_data;
c = c->next) {
- nchannels++;
- }
- channels = gpr_malloc(sizeof(channel_data *) * nchannels);
- i = 0;
- for (c = server->root_channel_data.next; c != &server->root_channel_data;
- c = c->next) {
- grpc_channel_internal_ref(c->channel);
- channels[i] = c;
- i++;
+ shutdown_channel(c, 1, c->num_calls == 0);
}
/* collect all unregistered then registered calls */
@@ -897,21 +943,6 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
maybe_finish_shutdown(server);
gpr_mu_unlock(&server->mu);
- for (i = 0; i < nchannels; i++) {
- c = channels[i];
- elem = grpc_channel_stack_element(
- grpc_channel_get_channel_stack(c->channel), 0);
-
- op.type = GRPC_CHANNEL_GOAWAY;
- op.dir = GRPC_CALL_DOWN;
- op.data.goaway.status = GRPC_STATUS_OK;
- op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown");
- elem->filter->channel_op(elem, NULL, &op);
-
- grpc_channel_internal_unref(c->channel);
- }
- gpr_free(channels);
-
/* terminate all the requested calls */
for (i = 0; i < requested_calls.count; i++) {
fail_call(server, &requested_calls.calls[i]);
@@ -953,7 +984,9 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
call_count = 0;
calls = gpr_malloc(sizeof(grpc_call *) * call_capacity);
- for (calld = server->lists[ALL_CALLS]; calld != server->lists[ALL_CALLS] || is_first; calld = calld->links[ALL_CALLS].next) {
+ for (calld = server->lists[ALL_CALLS];
+ calld != server->lists[ALL_CALLS] || is_first;
+ calld = calld->links[ALL_CALLS].next) {
if (call_count == call_capacity) {
call_capacity *= 2;
calls = gpr_realloc(calls, sizeof(grpc_call *) * call_capacity);
@@ -966,7 +999,8 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
gpr_mu_unlock(&server->mu);
for (i = 0; i < call_count; i++) {
- grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE, "Unavailable");
+ grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE,
+ "Unavailable");
GRPC_CALL_INTERNAL_UNREF(calls[i], "cancel_all", 1);
}
@@ -974,9 +1008,7 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
}
void grpc_server_destroy(grpc_server *server) {
- channel_data *c;
listener *l;
- call_data *calld;
gpr_mu_lock(&server->mu);
GPR_ASSERT(server->shutdown || !server->listeners);
@@ -988,19 +1020,6 @@ void grpc_server_destroy(grpc_server *server) {
gpr_free(l);
}
- while ((calld = call_list_remove_head(&server->lists[PENDING_START],
- PENDING_START)) != NULL) {
- calld->state = ZOMBIED;
- grpc_iomgr_closure_init(
- &calld->kill_zombie_closure, kill_zombie,
- grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
- grpc_iomgr_add_callback(&calld->kill_zombie_closure);
- }
-
- for (c = server->root_channel_data.next; c != &server->root_channel_data;
- c = c->next) {
- shutdown_channel(c);
- }
gpr_mu_unlock(&server->mu);
server_unref(server);
@@ -1063,6 +1082,9 @@ grpc_call_error grpc_server_request_call(
GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details,
initial_metadata, cq_bound_to_call,
cq_for_notification, tag);
+ if (!grpc_cq_is_server_cq(cq_for_notification)) {
+ return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+ }
grpc_cq_begin_op(cq_for_notification, NULL);
rc.type = BATCH_CALL;
rc.tag = tag;
@@ -1081,6 +1103,9 @@ grpc_call_error grpc_server_request_registered_call(
grpc_completion_queue *cq_for_notification, void *tag) {
requested_call rc;
registered_method *registered_method = rm;
+ if (!grpc_cq_is_server_cq(cq_for_notification)) {
+ return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+ }
grpc_cq_begin_op(cq_for_notification, NULL);
rc.type = REGISTERED_CALL;
rc.tag = tag;
@@ -1192,4 +1217,3 @@ int grpc_server_has_open_connections(grpc_server *server) {
gpr_mu_unlock(&server->mu);
return r;
}
-
generated by cgit v1.2.3 (git 2.39.1) at 2024-08-14 12:26:27 +0000