Chiebot-Mirror
/
grpc
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix end2end leaks

Browse Source
pull/501/head
Craig Tiller 10 years ago
parent 3d20c5e8c6
commit ec3257c120
5 changed files with 82 additions and 30 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      include/grpc++/impl/call.h
  2. 77
      src/core/surface/server.c
  3. 21
      src/cpp/common/call.cc
  4. 12
      src/cpp/server/server.cc
  5. 1
      test/cpp/end2end/end2end_test.cc

1
include/grpc++/impl/call.h
Unescape View File

@ -57,6 +57,7 @@ class Call;
class CallOpBuffer final : public CompletionQueueTag {
public:
CallOpBuffer() : return_tag_(this) {}
~CallOpBuffer();
void Reset(void *next_return_tag);

77
src/core/surface/server.c
Unescape View File

@ -53,7 +53,8 @@ typedef enum { PENDING_START, ALL_CALLS, CALL_LIST_COUNT } call_list;
typedef struct listener {
void *arg;
void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets, size_t pollset_count);
void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets,
size_t pollset_count);
void (*destroy)(grpc_server *server, void *arg);
struct listener *next;
} listener;
@ -129,7 +130,7 @@ struct grpc_server {
const grpc_channel_filter **channel_filters;
grpc_channel_args *channel_args;
grpc_completion_queue *unregistered_cq;
grpc_completion_queue **cqs;
grpc_pollset **pollsets;
size_t cq_count;
@ -257,11 +258,21 @@ static void server_ref(grpc_server *server) {
}
static void server_unref(grpc_server *server) {
registered_method *rm;
if (gpr_unref(&server->internal_refcount)) {
grpc_channel_args_destroy(server->channel_args);
gpr_mu_destroy(&server->mu);
gpr_free(server->channel_filters);
requested_call_array_destroy(&server->requested_calls);
while ((rm = server->registered_methods) != NULL) {
server->registered_methods = rm->next;
gpr_free(rm->method);
gpr_free(rm->host);
requested_call_array_destroy(&rm->requested);
gpr_free(rm);
}
gpr_free(server->cqs);
gpr_free(server->pollsets);
gpr_free(server);
}
}
@ -511,7 +522,8 @@ static void destroy_call_elem(grpc_call_element *elem) {
if (chand->server->shutdown && chand->server->have_shutdown_tag &&
chand->server->lists[ALL_CALLS] == NULL) {
for (i = 0; i < chand->server->cq_count; i++) {
grpc_cq_end_server_shutdown(chand->server->cqs[i], chand->server->shutdown_tag);
grpc_cq_end_server_shutdown(chand->server->cqs[i],
chand->server->shutdown_tag);
}
}
gpr_mu_unlock(&chand->server->mu);
@ -547,7 +559,19 @@ static void init_channel_elem(grpc_channel_element *elem,
}
static void destroy_channel_elem(grpc_channel_element *elem) {
size_t i;
channel_data *chand = elem->channel_data;
if (chand->registered_methods) {
for (i = 0; i < chand->registered_method_slots; i++) {
if (chand->registered_methods[i].method) {
grpc_mdstr_unref(chand->registered_methods[i].method);
}
if (chand->registered_methods[i].host) {
grpc_mdstr_unref(chand->registered_methods[i].host);
}
}
gpr_free(chand->registered_methods);
}
if (chand->server) {
gpr_mu_lock(&chand->server->mu);
chand->next->prev = chand->prev;
@ -571,7 +595,8 @@ static void addcq(grpc_server *server, grpc_completion_queue *cq) {
if (server->cqs[i] == cq) return;
}
n = server->cq_count++;
server->cqs = gpr_realloc(server->cqs, server->cq_count * sizeof(grpc_completion_queue*));
server->cqs = gpr_realloc(server->cqs,
server->cq_count * sizeof(grpc_completion_queue *));
server->cqs[n] = cq;
}
@ -624,7 +649,8 @@ static int streq(const char *a, const char *b) {
}
void *grpc_server_register_method(grpc_server *server, const char *method,
const char *host, grpc_completion_queue *cq_new_rpc) {
const char *host,
grpc_completion_queue *cq_new_rpc) {
registered_method *m;
if (!method) {
gpr_log(GPR_ERROR, "%s method string cannot be NULL", __FUNCTION__);
@ -652,7 +678,7 @@ void grpc_server_start(grpc_server *server) {
listener *l;
size_t i;
server->pollsets = gpr_malloc(sizeof(grpc_pollset*) * server->cq_count);
server->pollsets = gpr_malloc(sizeof(grpc_pollset *) * server->cq_count);
for (i = 0; i < server->cq_count; i++) {
server->pollsets[i] = grpc_cq_pollset(server->cqs[i]);
}
@ -745,7 +771,7 @@ grpc_transport_setup_result grpc_server_setup_transport(
}
static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
void *shutdown_tag) {
void *shutdown_tag) {
listener *l;
requested_call_array requested_calls;
channel_data **channels;
@ -781,12 +807,19 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
requested_calls = server->requested_calls;
memset(&server->requested_calls, 0, sizeof(server->requested_calls));
for (rm = server->registered_methods; rm; rm = rm->next) {
if (requested_calls.count + rm->requested.count > requested_calls.capacity) {
requested_calls.capacity = GPR_MAX(requested_calls.count + rm->requested.count, 2 * requested_calls.capacity);
requested_calls.calls = gpr_realloc(requested_calls.calls, sizeof(*requested_calls.calls) * requested_calls.capacity);
if (requested_calls.count + rm->requested.count >
requested_calls.capacity) {
requested_calls.capacity =
GPR_MAX(requested_calls.count + rm->requested.count,
2 * requested_calls.capacity);
requested_calls.calls =
gpr_realloc(requested_calls.calls, sizeof(*requested_calls.calls) *
requested_calls.capacity);
}
memcpy(requested_calls.calls + requested_calls.count, rm->requested.calls, sizeof(*requested_calls.calls) * rm->requested.count);
memcpy(requested_calls.calls + requested_calls.count, rm->requested.calls,
sizeof(*requested_calls.calls) * rm->requested.count);
requested_calls.count += rm->requested.count;
gpr_free(rm->requested.calls);
memset(&rm->requested, 0, sizeof(rm->requested));
}
@ -857,7 +890,8 @@ void grpc_server_destroy(grpc_server *server) {
void grpc_server_add_listener(grpc_server *server, void *arg,
void (*start)(grpc_server *server, void *arg,
grpc_pollset **pollsets, size_t pollset_count),
grpc_pollset **pollsets,
size_t pollset_count),
void (*destroy)(grpc_server *server, void *arg)) {
listener *l = gpr_malloc(sizeof(listener));
l->arg = arg;
@ -920,10 +954,9 @@ grpc_call_error grpc_server_request_call(grpc_server *server, grpc_call **call,
}
grpc_call_error grpc_server_request_registered_call(
grpc_server *server, void *rm, grpc_call **call,
gpr_timespec *deadline, grpc_metadata_array *initial_metadata,
grpc_byte_buffer **optional_payload, grpc_completion_queue *cq_bind,
void *tag) {
grpc_server *server, void *rm, grpc_call **call, gpr_timespec *deadline,
grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload,
grpc_completion_queue *cq_bind, void *tag) {
requested_call rc;
registered_method *registered_method = rm;
grpc_cq_begin_op(registered_method->cq, NULL, GRPC_OP_COMPLETE);
@ -1025,20 +1058,20 @@ static void begin_call(grpc_server *server, call_data *calld,
static void fail_call(grpc_server *server, requested_call *rc) {
switch (rc->type) {
case LEGACY_CALL:
grpc_cq_end_new_rpc(server->unregistered_cq, rc->tag, NULL, do_nothing, NULL, NULL,
NULL, gpr_inf_past, 0, NULL);
grpc_cq_end_new_rpc(server->unregistered_cq, rc->tag, NULL, do_nothing,
NULL, NULL, NULL, gpr_inf_past, 0, NULL);
break;
case BATCH_CALL:
*rc->data.batch.call = NULL;
rc->data.batch.initial_metadata->count = 0;
grpc_cq_end_op_complete(server->unregistered_cq, rc->tag, NULL, do_nothing, NULL,
GRPC_OP_ERROR);
grpc_cq_end_op_complete(server->unregistered_cq, rc->tag, NULL,
do_nothing, NULL, GRPC_OP_ERROR);
break;
case REGISTERED_CALL:
*rc->data.registered.call = NULL;
rc->data.registered.initial_metadata->count = 0;
grpc_cq_end_op_complete(rc->data.registered.registered_method->cq, rc->tag, NULL, do_nothing, NULL,
GRPC_OP_ERROR);
grpc_cq_end_op_complete(rc->data.registered.registered_method->cq,
rc->tag, NULL, do_nothing, NULL, GRPC_OP_ERROR);
break;
}
}

21
src/cpp/common/call.cc
Unescape View File

@ -48,8 +48,7 @@ void CallOpBuffer::Reset(void* next_return_tag) {
gpr_free(initial_metadata_);
recv_initial_metadata_ = nullptr;
gpr_free(recv_initial_metadata_arr_.metadata);
recv_initial_metadata_arr_ = {0, 0, nullptr};
recv_initial_metadata_arr_.count = 0;
send_message_ = nullptr;
if (send_message_buf_) {
@ -68,13 +67,9 @@ void CallOpBuffer::Reset(void* next_return_tag) {
recv_trailing_metadata_ = nullptr;
recv_status_ = nullptr;
gpr_free(recv_trailing_metadata_arr_.metadata);
recv_trailing_metadata_arr_ = {0, 0, nullptr};
recv_trailing_metadata_arr_.count = 0;
status_code_ = GRPC_STATUS_OK;
gpr_free(status_details_);
status_details_ = nullptr;
status_details_capacity_ = 0;
send_status_ = nullptr;
trailing_metadata_count_ = 0;
@ -83,6 +78,18 @@ void CallOpBuffer::Reset(void* next_return_tag) {
recv_closed_ = nullptr;
}
CallOpBuffer::~CallOpBuffer() {
gpr_free(status_details_);
gpr_free(recv_initial_metadata_arr_.metadata);
gpr_free(recv_trailing_metadata_arr_.metadata);
if (recv_message_buf_) {
grpc_byte_buffer_destroy(recv_message_buf_);
}
if (send_message_buf_) {
grpc_byte_buffer_destroy(send_message_buf_);
}
}
namespace {
// TODO(yangg) if the map is changed before we send, the pointers will be a
// mess. Make sure it does not happen.

12
src/cpp/server/server.cc
Unescape View File

@ -163,7 +163,11 @@ class Server::SyncRequest final : public CompletionQueueTag {
this));
}
void FinalizeResult(void** tag, bool* status) override {}
void FinalizeResult(void** tag, bool* status) override {
if (!*status) {
grpc_completion_queue_destroy(cq_);
}
}
class CallData final {
public:
@ -182,6 +186,12 @@ class Server::SyncRequest final : public CompletionQueueTag {
mrd->request_metadata_.count = 0;
}
~CallData() {
if (has_request_payload_ && request_payload_) {
grpc_byte_buffer_destroy(request_payload_);
}
}
void Run() {
std::unique_ptr<google::protobuf::Message> req;
std::unique_ptr<google::protobuf::Message> res;

1
test/cpp/end2end/end2end_test.cc
Unescape View File

@ -445,5 +445,6 @@ int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
int result = RUN_ALL_TESTS();
grpc_shutdown();
google::protobuf::ShutdownProtobufLibrary();
return result;
}

Write Preview
Loading…
Cancel
Save