diff options
Diffstat (limited to 'src/core/lib')
76 files changed, 1629 insertions, 875 deletions
diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c index 3fb2a60ac7..6d53b0576e 100644 --- a/src/core/lib/channel/channel_stack.c +++ b/src/core/lib/channel/channel_stack.c @@ -166,41 +166,32 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx, } } -grpc_error *grpc_call_stack_init( - grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack, - int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg, - grpc_call_context_element *context, const void *transport_server_data, - grpc_slice path, gpr_timespec start_time, gpr_timespec deadline, - grpc_call_stack *call_stack) { +grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx, + grpc_channel_stack *channel_stack, + int initial_refs, grpc_iomgr_cb_func destroy, + void *destroy_arg, + const grpc_call_element_args *elem_args) { grpc_channel_element *channel_elems = CHANNEL_ELEMS_FROM_STACK(channel_stack); size_t count = channel_stack->count; grpc_call_element *call_elems; char *user_data; size_t i; - call_stack->count = count; - GRPC_STREAM_REF_INIT(&call_stack->refcount, initial_refs, destroy, + elem_args->call_stack->count = count; + GRPC_STREAM_REF_INIT(&elem_args->call_stack->refcount, initial_refs, destroy, destroy_arg, "CALL_STACK"); - call_elems = CALL_ELEMS_FROM_STACK(call_stack); + call_elems = CALL_ELEMS_FROM_STACK(elem_args->call_stack); user_data = ((char *)call_elems) + ROUND_UP_TO_ALIGNMENT_SIZE(count * sizeof(grpc_call_element)); /* init per-filter data */ grpc_error *first_error = GRPC_ERROR_NONE; - const grpc_call_element_args args = { - .start_time = start_time, - .call_stack = call_stack, - .server_transport_data = transport_server_data, - .context = context, - .path = path, - .deadline = deadline, - }; for (i = 0; i < count; i++) { call_elems[i].filter = channel_elems[i].filter; call_elems[i].channel_data = channel_elems[i].channel_data; call_elems[i].call_data = user_data; - grpc_error *error = - call_elems[i].filter->init_call_elem(exec_ctx, &call_elems[i], &args); + grpc_error *error = call_elems[i].filter->init_call_elem( + exec_ctx, &call_elems[i], elem_args); if (error != GRPC_ERROR_NONE) { if (first_error == GRPC_ERROR_NONE) { first_error = error; @@ -241,15 +232,16 @@ void grpc_call_stack_ignore_set_pollset_or_pollset_set( void grpc_call_stack_destroy(grpc_exec_ctx *exec_ctx, grpc_call_stack *stack, const grpc_call_final_info *final_info, - void *and_free_memory) { + grpc_closure *then_schedule_closure) { grpc_call_element *elems = CALL_ELEMS_FROM_STACK(stack); size_t count = stack->count; size_t i; /* destroy per-filter data */ for (i = 0; i < count; i++) { - elems[i].filter->destroy_call_elem(exec_ctx, &elems[i], final_info, - i == count - 1 ? and_free_memory : NULL); + elems[i].filter->destroy_call_elem( + exec_ctx, &elems[i], final_info, + i == count - 1 ? then_schedule_closure : NULL); } } diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h index 1e943dc2e5..80e3603e8d 100644 --- a/src/core/lib/channel/channel_stack.h +++ b/src/core/lib/channel/channel_stack.h @@ -56,6 +56,7 @@ #include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/polling_entity.h" +#include "src/core/lib/support/arena.h" #include "src/core/lib/transport/transport.h" #ifdef __cplusplus @@ -84,6 +85,7 @@ typedef struct { grpc_slice path; gpr_timespec start_time; gpr_timespec deadline; + gpr_arena *arena; } grpc_call_element_args; typedef struct { @@ -128,7 +130,8 @@ typedef struct { server_transport_data is an opaque pointer. If it is NULL, this call is on a client; if it is non-NULL, then it points to memory owned by the transport and is on the server. Most filters want to ignore this - argument. */ + argument. + Implementations may assume that elem->call_data is all zeros. */ grpc_error *(*init_call_elem)(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_element_args *args); @@ -138,12 +141,12 @@ typedef struct { /* Destroy per call data. The filter does not need to do any chaining. The bottom filter of a stack will be passed a non-NULL pointer to - \a and_free_memory that should be passed to gpr_free when destruction - is complete. \a final_info contains data about the completed call, mainly - for reporting purposes. */ + \a then_schedule_closure that should be passed to grpc_closure_sched when + destruction is complete. \a final_info contains data about the completed + call, mainly for reporting purposes. */ void (*destroy_call_elem)(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory); + grpc_closure *then_schedule_closure); /* sizeof(per channel data) */ size_t sizeof_channel_data; @@ -152,7 +155,8 @@ typedef struct { is what needs initializing. is_first, is_last designate this elements position in the stack, and are useful for asserting correct configuration by upper layer code. - The filter does not need to do any chaining */ + The filter does not need to do any chaining. + Implementations may assume that elem->call_data is all zeros. */ grpc_error *(*init_channel_elem)(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_channel_element_args *args); @@ -234,12 +238,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx, /* Initialize a call stack given a channel stack. transport_server_data is expected to be NULL on a client, or an opaque transport owned pointer on the server. */ -grpc_error *grpc_call_stack_init( - grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack, - int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg, - grpc_call_context_element *context, const void *transport_server_data, - grpc_slice path, gpr_timespec start_time, gpr_timespec deadline, - grpc_call_stack *call_stack); +grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx, + grpc_channel_stack *channel_stack, + int initial_refs, grpc_iomgr_cb_func destroy, + void *destroy_arg, + const grpc_call_element_args *elem_args); /* Set a pollset or a pollset_set for a call stack: must occur before the first * op is started */ void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, @@ -269,7 +272,7 @@ void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, /* Destroy a call stack */ void grpc_call_stack_destroy(grpc_exec_ctx *exec_ctx, grpc_call_stack *stack, const grpc_call_final_info *final_info, - void *and_free_memory); + grpc_closure *then_schedule_closure); /* Ignore set pollset{_set} - used by filters if they don't care about pollsets * at all. Does nothing. */ diff --git a/src/core/lib/channel/channel_stack_builder.c b/src/core/lib/channel/channel_stack_builder.c index 5f9e3b4539..b515b7321a 100644 --- a/src/core/lib/channel/channel_stack_builder.c +++ b/src/core/lib/channel/channel_stack_builder.c @@ -65,8 +65,7 @@ struct grpc_channel_stack_builder_iterator { }; grpc_channel_stack_builder *grpc_channel_stack_builder_create(void) { - grpc_channel_stack_builder *b = gpr_malloc(sizeof(*b)); - memset(b, 0, sizeof(*b)); + grpc_channel_stack_builder *b = gpr_zalloc(sizeof(*b)); b->begin.filter = NULL; b->end.filter = NULL; @@ -251,7 +250,7 @@ grpc_error *grpc_channel_stack_builder_finish( size_t channel_stack_size = grpc_channel_stack_size(filters, num_filters); // allocate memory, with prefix_bytes followed by channel_stack_size - *result = gpr_malloc(prefix_bytes + channel_stack_size); + *result = gpr_zalloc(prefix_bytes + channel_stack_size); // fetch a pointer to the channel stack grpc_channel_stack *channel_stack = (grpc_channel_stack *)((char *)(*result) + prefix_bytes); diff --git a/src/core/lib/channel/compress_filter.c b/src/core/lib/channel/compress_filter.c index aa41014a21..02dc479f3a 100644 --- a/src/core/lib/channel/compress_filter.c +++ b/src/core/lib/channel/compress_filter.c @@ -292,7 +292,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *ignored) { + grpc_closure *ignored) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices); diff --git a/src/core/lib/channel/connected_channel.c b/src/core/lib/channel/connected_channel.c index 29796f7ca7..42ef7b7806 100644 --- a/src/core/lib/channel/connected_channel.c +++ b/src/core/lib/channel/connected_channel.c @@ -88,7 +88,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, channel_data *chand = elem->channel_data; int r = grpc_transport_init_stream( exec_ctx, chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld), - &args->call_stack->refcount, args->server_transport_data); + &args->call_stack->refcount, args->server_transport_data, args->arena); return r == 0 ? GRPC_ERROR_NONE : GRPC_ERROR_CREATE("transport stream initialization failed"); } @@ -105,12 +105,12 @@ static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory) { + grpc_closure *then_schedule_closure) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; grpc_transport_destroy_stream(exec_ctx, chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld), - and_free_memory); + then_schedule_closure); } /* Constructor for channel_data */ diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c index f9668be0fa..34114bbebf 100644 --- a/src/core/lib/channel/deadline_filter.c +++ b/src/core/lib/channel/deadline_filter.c @@ -144,7 +144,6 @@ static void inject_on_complete_cb(grpc_deadline_state* deadline_state, void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, grpc_call_stack* call_stack) { grpc_deadline_state* deadline_state = elem->call_data; - memset(deadline_state, 0, sizeof(*deadline_state)); deadline_state->call_stack = call_stack; } @@ -249,8 +248,6 @@ typedef struct server_call_data { static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, const grpc_call_element_args* args) { - // Note: size of call data is different between client and server. - memset(elem->call_data, 0, elem->filter->sizeof_call_data); grpc_deadline_state_init(exec_ctx, elem, args->call_stack); grpc_deadline_state_start(exec_ctx, elem, args->deadline); return GRPC_ERROR_NONE; @@ -259,7 +256,7 @@ static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, // Destructor for call_data. Used for both client and server filters. static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, const grpc_call_final_info* final_info, - void* and_free_memory) { + grpc_closure* ignored) { grpc_deadline_state_destroy(exec_ctx, elem); } diff --git a/src/core/lib/channel/deadline_filter.h b/src/core/lib/channel/deadline_filter.h index 94717f6bc7..72cd5cb929 100644 --- a/src/core/lib/channel/deadline_filter.h +++ b/src/core/lib/channel/deadline_filter.h @@ -62,6 +62,7 @@ typedef struct grpc_deadline_state { // elem->call_data is a grpc_deadline_state. // +// assumes elem->call_data is zero'd void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, grpc_call_stack* call_stack); void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx, diff --git a/src/core/lib/channel/handshaker.c b/src/core/lib/channel/handshaker.c index 82c361c7ef..1b4240bb10 100644 --- a/src/core/lib/channel/handshaker.c +++ b/src/core/lib/channel/handshaker.c @@ -99,8 +99,7 @@ struct grpc_handshake_manager { }; grpc_handshake_manager* grpc_handshake_manager_create() { - grpc_handshake_manager* mgr = gpr_malloc(sizeof(grpc_handshake_manager)); - memset(mgr, 0, sizeof(*mgr)); + grpc_handshake_manager* mgr = gpr_zalloc(sizeof(grpc_handshake_manager)); gpr_mu_init(&mgr->mu); gpr_ref_init(&mgr->refs, 1); return mgr; diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c index c031533dd8..f9d0d689ac 100644 --- a/src/core/lib/channel/http_client_filter.c +++ b/src/core/lib/channel/http_client_filter.c @@ -412,7 +412,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *ignored) { + grpc_closure *ignored) { call_data *calld = elem->call_data; grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices); } diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c index ce519f9c92..bebd3af335 100644 --- a/src/core/lib/channel/http_server_filter.c +++ b/src/core/lib/channel/http_server_filter.c @@ -345,7 +345,6 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; /* initialize members */ - memset(calld, 0, sizeof(*calld)); grpc_closure_init(&calld->hs_on_recv, hs_on_recv, elem, grpc_schedule_on_exec_ctx); grpc_closure_init(&calld->hs_on_complete, hs_on_complete, elem, @@ -359,7 +358,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *ignored) { + grpc_closure *ignored) { call_data *calld = elem->call_data; grpc_slice_buffer_destroy_internal(exec_ctx, &calld->read_slice_buffer); } diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c index 22938c64b5..5ba13fe251 100644 --- a/src/core/lib/channel/message_size_filter.c +++ b/src/core/lib/channel/message_size_filter.c @@ -200,7 +200,7 @@ static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, // Destructor for call_data. static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, const grpc_call_final_info* final_info, - void* ignored) {} + grpc_closure* ignored) {} // Constructor for channel_data. static grpc_error* init_channel_elem(grpc_exec_ctx* exec_ctx, @@ -208,7 +208,6 @@ static grpc_error* init_channel_elem(grpc_exec_ctx* exec_ctx, grpc_channel_element_args* args) { GPR_ASSERT(!args->is_last); channel_data* chand = elem->channel_data; - memset(chand, 0, sizeof(*chand)); chand->max_send_size = GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH; chand->max_recv_size = GRPC_DEFAULT_MAX_RECV_MESSAGE_LENGTH; for (size_t i = 0; i < args->channel_args->num_args; ++i) { diff --git a/src/core/lib/http/httpcli_security_connector.c b/src/core/lib/http/httpcli_security_connector.c index f4f6f3c27a..354d2f4a09 100644 --- a/src/core/lib/http/httpcli_security_connector.c +++ b/src/core/lib/http/httpcli_security_connector.c @@ -118,8 +118,7 @@ static grpc_security_status httpcli_ssl_channel_security_connector_create( return GRPC_SECURITY_ERROR; } - c = gpr_malloc(sizeof(grpc_httpcli_ssl_channel_security_connector)); - memset(c, 0, sizeof(grpc_httpcli_ssl_channel_security_connector)); + c = gpr_zalloc(sizeof(grpc_httpcli_ssl_channel_security_connector)); gpr_ref_init(&c->base.base.refcount, 1); c->base.base.vtable = &httpcli_ssl_vtable; diff --git a/src/core/lib/http/parser.c b/src/core/lib/http/parser.c index 2f84adc187..b9c56c103c 100644 --- a/src/core/lib/http/parser.c +++ b/src/core/lib/http/parser.c @@ -284,9 +284,9 @@ static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte, case GRPC_HTTP_HEADERS: if (parser->cur_line_length >= GRPC_HTTP_PARSER_MAX_HEADER_LENGTH) { if (grpc_http1_trace) - gpr_log(GPR_ERROR, "HTTP client max line length (%d) exceeded", + gpr_log(GPR_ERROR, "HTTP header max line length (%d) exceeded", GRPC_HTTP_PARSER_MAX_HEADER_LENGTH); - return GRPC_ERROR_NONE; + return GRPC_ERROR_CREATE("HTTP header max line length exceeded"); } parser->cur_line[parser->cur_line_length] = byte; parser->cur_line_length++; diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c index dbe5b139f9..1127fff756 100644 --- a/src/core/lib/iomgr/error.c +++ b/src/core/lib/iomgr/error.c @@ -35,6 +35,7 @@ #include <string.h> +#include <grpc/slice.h> #include <grpc/status.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> @@ -47,46 +48,7 @@ #include "src/core/lib/iomgr/error_internal.h" #include "src/core/lib/profiling/timers.h" - -static void destroy_integer(void *key) {} - -static void *copy_integer(void *key) { return key; } - -static long compare_integers(void *key1, void *key2) { - return GPR_ICMP((uintptr_t)key1, (uintptr_t)key2); -} - -static void destroy_string(void *str) { gpr_free(str); } - -static void *copy_string(void *str) { return gpr_strdup(str); } - -static void destroy_err(void *err) { GRPC_ERROR_UNREF(err); } - -static void *copy_err(void *err) { return GRPC_ERROR_REF(err); } - -static void destroy_time(void *tm) { gpr_free(tm); } - -static gpr_timespec *box_time(gpr_timespec tm) { - gpr_timespec *out = gpr_malloc(sizeof(*out)); - *out = tm; - return out; -} - -static void *copy_time(void *tm) { return box_time(*(gpr_timespec *)tm); } - -static const gpr_avl_vtable avl_vtable_ints = {destroy_integer, copy_integer, - compare_integers, - destroy_integer, copy_integer}; - -static const gpr_avl_vtable avl_vtable_strs = {destroy_integer, copy_integer, - compare_integers, destroy_string, - copy_string}; - -static const gpr_avl_vtable avl_vtable_times = { - destroy_integer, copy_integer, compare_integers, destroy_time, copy_time}; - -static const gpr_avl_vtable avl_vtable_errs = { - destroy_integer, copy_integer, compare_integers, destroy_err, copy_err}; +#include "src/core/lib/slice/slice_internal.h" static const char *error_int_name(grpc_error_ints key) { switch (key) { @@ -120,6 +82,8 @@ static const char *error_int_name(grpc_error_ints key) { return "limit"; case GRPC_ERROR_INT_OCCURRED_DURING_WRITE: return "occurred_during_write"; + case GRPC_ERROR_INT_MAX: + GPR_UNREACHABLE_CODE(return "unknown"); } GPR_UNREACHABLE_CODE(return "unknown"); } @@ -150,6 +114,8 @@ static const char *error_str_name(grpc_error_strs key) { return "filename"; case GRPC_ERROR_STR_QUEUED_BUFFERS: return "queued_buffers"; + case GRPC_ERROR_STR_MAX: + GPR_UNREACHABLE_CODE(return "unknown"); } GPR_UNREACHABLE_CODE(return "unknown"); } @@ -158,6 +124,8 @@ static const char *error_time_name(grpc_error_times key) { switch (key) { case GRPC_ERROR_TIME_CREATED: return "created"; + case GRPC_ERROR_TIME_MAX: + GPR_UNREACHABLE_CODE(return "unknown"); } GPR_UNREACHABLE_CODE(return "unknown"); } @@ -172,25 +140,51 @@ grpc_error *grpc_error_ref(grpc_error *err, const char *file, int line, const char *func) { if (grpc_error_is_special(err)) return err; gpr_log(GPR_DEBUG, "%p: %" PRIdPTR " -> %" PRIdPTR " [%s:%d %s]", err, - err->refs.count, err->refs.count + 1, file, line, func); - gpr_ref(&err->refs); + gpr_atm_no_barrier_load(&err->atomics.refs.count), + gpr_atm_no_barrier_load(&err->atomics.refs.count) + 1, file, line, + func); + gpr_ref(&err->atomics.refs); return err; } #else grpc_error *grpc_error_ref(grpc_error *err) { if (grpc_error_is_special(err)) return err; - gpr_ref(&err->refs); + gpr_ref(&err->atomics.refs); return err; } #endif +static void unref_errs(grpc_error *err) { + uint8_t slot = err->first_err; + while (slot != UINT8_MAX) { + grpc_linked_error *lerr = (grpc_linked_error *)(err->arena + slot); + GRPC_ERROR_UNREF(lerr->err); + GPR_ASSERT(err->last_err == slot ? lerr->next == UINT8_MAX + : lerr->next != UINT8_MAX); + slot = lerr->next; + } +} + +static void unref_slice(grpc_slice slice) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_slice_unref_internal(&exec_ctx, slice); + grpc_exec_ctx_finish(&exec_ctx); +} + +static void unref_strs(grpc_error *err) { + for (size_t which = 0; which < GRPC_ERROR_STR_MAX; ++which) { + uint8_t slot = err->strs[which]; + if (slot != UINT8_MAX) { + unref_slice(*(grpc_slice *)(err->arena + slot)); + } + } +} + static void error_destroy(grpc_error *err) { GPR_ASSERT(!grpc_error_is_special(err)); - gpr_avl_unref(err->ints); - gpr_avl_unref(err->strs); - gpr_avl_unref(err->errs); - gpr_avl_unref(err->times); - gpr_free((void *)gpr_atm_acq_load(&err->error_string)); + unref_errs(err); + unref_strs(err); + gpr_free((void *)gpr_atm_acq_load(&err->atomics.error_string)); gpr_free(err); } @@ -199,81 +193,209 @@ void grpc_error_unref(grpc_error *err, const char *file, int line, const char *func) { if (grpc_error_is_special(err)) return; gpr_log(GPR_DEBUG, "%p: %" PRIdPTR " -> %" PRIdPTR " [%s:%d %s]", err, - err->refs.count, err->refs.count - 1, file, line, func); - if (gpr_unref(&err->refs)) { + gpr_atm_no_barrier_load(&err->atomics.refs.count), + gpr_atm_no_barrier_load(&err->atomics.refs.count) - 1, file, line, + func); + if (gpr_unref(&err->atomics.refs)) { error_destroy(err); } } #else void grpc_error_unref(grpc_error *err) { if (grpc_error_is_special(err)) return; - if (gpr_unref(&err->refs)) { + if (gpr_unref(&err->atomics.refs)) { error_destroy(err); } } #endif +static uint8_t get_placement(grpc_error **err, size_t size) { + GPR_ASSERT(*err); + uint8_t slots = (uint8_t)(size / sizeof(intptr_t)); + if ((*err)->arena_size + slots > (*err)->arena_capacity) { + (*err)->arena_capacity = (uint8_t)(3 * (*err)->arena_capacity / 2); + *err = gpr_realloc( + *err, sizeof(grpc_error) + (*err)->arena_capacity * sizeof(intptr_t)); + } + uint8_t placement = (*err)->arena_size; + (*err)->arena_size = (uint8_t)((*err)->arena_size + slots); + return placement; +} + +static void internal_set_int(grpc_error **err, grpc_error_ints which, + intptr_t value) { + // GPR_ASSERT((*err)->ints[which] == UINT8_MAX); // TODO, enforce this + uint8_t slot = (*err)->ints[which]; + if (slot == UINT8_MAX) { + slot = get_placement(err, sizeof(value)); + } + (*err)->ints[which] = slot; + (*err)->arena[slot] = value; +} + +static void internal_set_str(grpc_error **err, grpc_error_strs which, + grpc_slice value) { + // GPR_ASSERT((*err)->strs[which] == UINT8_MAX); // TODO, enforce this + uint8_t slot = (*err)->strs[which]; + if (slot == UINT8_MAX) { + slot = get_placement(err, sizeof(value)); + } else { + unref_slice(*(grpc_slice *)((*err)->arena + slot)); + } + (*err)->strs[which] = slot; + memcpy((*err)->arena + slot, &value, sizeof(value)); +} + +static void internal_set_time(grpc_error **err, grpc_error_times which, + gpr_timespec value) { + // GPR_ASSERT((*err)->times[which] == UINT8_MAX); // TODO, enforce this + uint8_t slot = (*err)->times[which]; + if (slot == UINT8_MAX) { + slot = get_placement(err, sizeof(value)); + } + (*err)->times[which] = slot; + memcpy((*err)->arena + slot, &value, sizeof(value)); +} + +static void internal_add_error(grpc_error **err, grpc_error *new) { + grpc_linked_error new_last = {new, UINT8_MAX}; + uint8_t slot = get_placement(err, sizeof(grpc_linked_error)); + if ((*err)->first_err == UINT8_MAX) { + GPR_ASSERT((*err)->last_err == UINT8_MAX); + (*err)->last_err = slot; + (*err)->first_err = slot; + } else { + GPR_ASSERT((*err)->last_err != UINT8_MAX); + grpc_linked_error *old_last = + (grpc_linked_error *)((*err)->arena + (*err)->last_err); + old_last->next = slot; + (*err)->last_err = slot; + } + memcpy((*err)->arena + slot, &new_last, sizeof(grpc_linked_error)); +} + +#define SLOTS_PER_INT (sizeof(intptr_t) / sizeof(intptr_t)) +#define SLOTS_PER_STR (sizeof(grpc_slice) / sizeof(intptr_t)) +#define SLOTS_PER_TIME (sizeof(gpr_timespec) / sizeof(intptr_t)) +#define SLOTS_PER_LINKED_ERROR (sizeof(grpc_linked_error) / sizeof(intptr_t)) + +// size of storing one int and two slices and a timespec. For line, desc, file, +// and time created +#define DEFAULT_ERROR_CAPACITY \ + (SLOTS_PER_INT + (SLOTS_PER_STR * 2) + SLOTS_PER_TIME) + +// It is very common to include and extra int and string in an error +#define SURPLUS_CAPACITY (2 * SLOTS_PER_INT + SLOTS_PER_TIME) + grpc_error *grpc_error_create(const char *file, int line, const char *desc, grpc_error **referencing, size_t num_referencing) { GPR_TIMER_BEGIN("grpc_error_create", 0); - grpc_error *err = gpr_malloc(sizeof(*err)); + uint8_t initial_arena_capacity = (uint8_t)( + DEFAULT_ERROR_CAPACITY + + (uint8_t)(num_referencing * SLOTS_PER_LINKED_ERROR) + SURPLUS_CAPACITY); + grpc_error *err = + gpr_malloc(sizeof(*err) + initial_arena_capacity * sizeof(intptr_t)); if (err == NULL) { // TODO(ctiller): make gpr_malloc return NULL return GRPC_ERROR_OOM; } #ifdef GRPC_ERROR_REFCOUNT_DEBUG gpr_log(GPR_DEBUG, "%p create [%s:%d]", err, file, line); #endif - err->ints = gpr_avl_add(gpr_avl_create(&avl_vtable_ints), - (void *)(uintptr_t)GRPC_ERROR_INT_FILE_LINE, - (void *)(uintptr_t)line); - err->strs = gpr_avl_add( - gpr_avl_add(gpr_avl_create(&avl_vtable_strs), - (void *)(uintptr_t)GRPC_ERROR_STR_FILE, gpr_strdup(file)), - (void *)(uintptr_t)GRPC_ERROR_STR_DESCRIPTION, gpr_strdup(desc)); - err->errs = gpr_avl_create(&avl_vtable_errs); - err->next_err = 0; - for (size_t i = 0; i < num_referencing; i++) { + + err->arena_size = 0; + err->arena_capacity = initial_arena_capacity; + err->first_err = UINT8_MAX; + err->last_err = UINT8_MAX; + + memset(err->ints, UINT8_MAX, GRPC_ERROR_INT_MAX); + memset(err->strs, UINT8_MAX, GRPC_ERROR_STR_MAX); + memset(err->times, UINT8_MAX, GRPC_ERROR_TIME_MAX); + + internal_set_int(&err, GRPC_ERROR_INT_FILE_LINE, line); + internal_set_str(&err, GRPC_ERROR_STR_FILE, + grpc_slice_from_static_string(file)); + internal_set_str( + &err, GRPC_ERROR_STR_DESCRIPTION, + grpc_slice_from_copied_buffer( + desc, + strlen(desc) + + 1)); // TODO, pull this up. // TODO(ncteisen), pull this up. + + for (size_t i = 0; i < num_referencing; ++i) { if (referencing[i] == GRPC_ERROR_NONE) continue; - err->errs = gpr_avl_add(err->errs, (void *)(err->next_err++), - GRPC_ERROR_REF(referencing[i])); - } - err->times = gpr_avl_add(gpr_avl_create(&avl_vtable_times), - (void *)(uintptr_t)GRPC_ERROR_TIME_CREATED, - box_time(gpr_now(GPR_CLOCK_REALTIME))); - gpr_atm_no_barrier_store(&err->error_string, 0); - gpr_ref_init(&err->refs, 1); + internal_add_error( + &err, + GRPC_ERROR_REF( + referencing[i])); // TODO(ncteisen), change ownership semantics + } + + internal_set_time(&err, GRPC_ERROR_TIME_CREATED, gpr_now(GPR_CLOCK_REALTIME)); + + gpr_atm_no_barrier_store(&err->atomics.error_string, 0); + gpr_ref_init(&err->atomics.refs, 1); GPR_TIMER_END("grpc_error_create", 0); return err; } +static void ref_strs(grpc_error *err) { + for (size_t i = 0; i < GRPC_ERROR_STR_MAX; ++i) { + uint8_t slot = err->strs[i]; + if (slot != UINT8_MAX) { + grpc_slice_ref_internal(*(grpc_slice *)(err->arena + slot)); + } + } +} + +static void ref_errs(grpc_error *err) { + uint8_t slot = err->first_err; + while (slot != UINT8_MAX) { + grpc_linked_error *lerr = (grpc_linked_error *)(err->arena + slot); + GRPC_ERROR_REF(lerr->err); + slot = lerr->next; + } +} + static grpc_error *copy_error_and_unref(grpc_error *in) { GPR_TIMER_BEGIN("copy_error_and_unref", 0); grpc_error *out; if (grpc_error_is_special(in)) { - if (in == GRPC_ERROR_NONE) - out = grpc_error_set_int(GRPC_ERROR_CREATE("no error"), - GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_OK); - else if (in == GRPC_ERROR_OOM) - out = GRPC_ERROR_CREATE("oom"); - else if (in == GRPC_ERROR_CANCELLED) - out = - grpc_error_set_int(GRPC_ERROR_CREATE("cancelled"), - GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_CANCELLED); - else - out = GRPC_ERROR_CREATE("unknown"); + out = GRPC_ERROR_CREATE("unknown"); + if (in == GRPC_ERROR_NONE) { + internal_set_str(&out, GRPC_ERROR_STR_DESCRIPTION, + grpc_slice_from_static_string("no error")); + internal_set_int(&out, GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_OK); + } else if (in == GRPC_ERROR_OOM) { + internal_set_str(&out, GRPC_ERROR_STR_DESCRIPTION, + grpc_slice_from_static_string("oom")); + } else if (in == GRPC_ERROR_CANCELLED) { + internal_set_str(&out, GRPC_ERROR_STR_DESCRIPTION, + grpc_slice_from_static_string("cancelled")); + internal_set_int(&out, GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_CANCELLED); + } + } else if (gpr_ref_is_unique(&in->atomics.refs)) { + out = in; } else { - out = gpr_malloc(sizeof(*out)); + uint8_t new_arena_capacity = in->arena_capacity; + // the returned err will be added to, so we ensure this is room to avoid + // unneeded allocations. + if (in->arena_capacity - in->arena_size < (uint8_t)SLOTS_PER_STR) { + new_arena_capacity = (uint8_t)(3 * new_arena_capacity / 2); + } + out = gpr_malloc(sizeof(*in) + new_arena_capacity * sizeof(intptr_t)); #ifdef GRPC_ERROR_REFCOUNT_DEBUG gpr_log(GPR_DEBUG, "%p create copying %p", out, in); #endif - out->ints = gpr_avl_ref(in->ints); - out->strs = gpr_avl_ref(in->strs); - out->errs = gpr_avl_ref(in->errs); - out->times = gpr_avl_ref(in->times); - gpr_atm_no_barrier_store(&out->error_string, 0); - out->next_err = in->next_err; - gpr_ref_init(&out->refs, 1); + // bulk memcpy of the rest of the struct. + size_t skip = sizeof(&out->atomics); + memcpy((void *)((uintptr_t)out + skip), (void *)((uintptr_t)in + skip), + sizeof(*in) + (in->arena_size * sizeof(intptr_t)) - skip); + // manually set the atomics and the new capacity + gpr_atm_no_barrier_store(&out->atomics.error_string, 0); + gpr_ref_init(&out->atomics.refs, 1); + out->arena_capacity = new_arena_capacity; + ref_strs(out); + ref_errs(out); GRPC_ERROR_UNREF(in); } GPR_TIMER_END("copy_error_and_unref", 0); @@ -284,7 +406,7 @@ grpc_error *grpc_error_set_int(grpc_error *src, grpc_error_ints which, intptr_t value) { GPR_TIMER_BEGIN("grpc_error_set_int", 0); grpc_error *new = copy_error_and_unref(src); - new->ints = gpr_avl_add(new->ints, (void *)(uintptr_t)which, (void *)value); + internal_set_int(&new, which, value); GPR_TIMER_END("grpc_error_set_int", 0); return new; } @@ -302,7 +424,6 @@ static special_error_status_map error_status_map[] = { bool grpc_error_get_int(grpc_error *err, grpc_error_ints which, intptr_t *p) { GPR_TIMER_BEGIN("grpc_error_get_int", 0); - void *pp; if (grpc_error_is_special(err)) { if (which == GRPC_ERROR_INT_GRPC_STATUS) { for (size_t i = 0; i < GPR_ARRAY_SIZE(error_status_map); i++) { @@ -316,8 +437,9 @@ bool grpc_error_get_int(grpc_error *err, grpc_error_ints which, intptr_t *p) { GPR_TIMER_END("grpc_error_get_int", 0); return false; } - if (gpr_avl_maybe_get(err->ints, (void *)(uintptr_t)which, &pp)) { - if (p != NULL) *p = (intptr_t)pp; + uint8_t slot = err->ints[which]; + if (slot != UINT8_MAX) { + if (p != NULL) *p = err->arena[slot]; GPR_TIMER_END("grpc_error_get_int", 0); return true; } @@ -329,8 +451,9 @@ grpc_error *grpc_error_set_str(grpc_error *src, grpc_error_strs which, const char *value) { GPR_TIMER_BEGIN("grpc_error_set_str", 0); grpc_error *new = copy_error_and_unref(src); - new->strs = - gpr_avl_add(new->strs, (void *)(uintptr_t)which, gpr_strdup(value)); + internal_set_str(&new, which, + grpc_slice_from_copied_buffer( + value, strlen(value) + 1)); // TODO, pull this up. GPR_TIMER_END("grpc_error_set_str", 0); return new; } @@ -346,13 +469,19 @@ const char *grpc_error_get_str(grpc_error *err, grpc_error_strs which) { } return NULL; } - return gpr_avl_get(err->strs, (void *)(uintptr_t)which); + uint8_t slot = err->strs[which]; + if (slot != UINT8_MAX) { + return (const char *)GRPC_SLICE_START_PTR( + *(grpc_slice *)(err->arena + slot)); + } else { + return NULL; + } } grpc_error *grpc_error_add_child(grpc_error *src, grpc_error *child) { GPR_TIMER_BEGIN("grpc_error_add_child", 0); grpc_error *new = copy_error_and_unref(src); - new->errs = gpr_avl_add(new->errs, (void *)(new->next_err++), child); + internal_add_error(&new, child); GPR_TIMER_END("grpc_error_add_child", 0); return new; } @@ -372,42 +501,6 @@ typedef struct { size_t cap_kvs; } kv_pairs; -static void append_kv(kv_pairs *kvs, char *key, char *value) { - if (kvs->num_kvs == kvs->cap_kvs) { - kvs->cap_kvs = GPR_MAX(3 * kvs->cap_kvs / 2, 4); - kvs->kvs = gpr_realloc(kvs->kvs, sizeof(*kvs->kvs) * kvs->cap_kvs); - } - kvs->kvs[kvs->num_kvs].key = key; - kvs->kvs[kvs->num_kvs].value = value; - kvs->num_kvs++; -} - -static void collect_kvs(gpr_avl_node *node, char *key(void *k), - char *fmt(void *v), kv_pairs *kvs) { - if (node == NULL) return; - append_kv(kvs, key(node->key), fmt(node->value)); - collect_kvs(node->left, key, fmt, kvs); - collect_kvs(node->right, key, fmt, kvs); -} - -static char *key_int(void *p) { - return gpr_strdup(error_int_name((grpc_error_ints)(uintptr_t)p)); -} - -static char *key_str(void *p) { - return gpr_strdup(error_str_name((grpc_error_strs)(uintptr_t)p)); -} - -static char *key_time(void *p) { - return gpr_strdup(error_time_name((grpc_error_times)(uintptr_t)p)); -} - -static char *fmt_int(void *p) { - char *s; - gpr_asprintf(&s, "%" PRIdPTR, (intptr_t)p); - return s; -} - static void append_chr(char c, char **s, size_t *sz, size_t *cap) { if (*sz == *cap) { *cap = GPR_MAX(8, 3 * *cap / 2); @@ -459,6 +552,40 @@ static void append_esc_str(const char *str, char **s, size_t *sz, size_t *cap) { append_chr('"', s, sz, cap); } +static void append_kv(kv_pairs *kvs, char *key, char *value) { + if (kvs->num_kvs == kvs->cap_kvs) { + kvs->cap_kvs = GPR_MAX(3 * kvs->cap_kvs / 2, 4); + kvs->kvs = gpr_realloc(kvs->kvs, sizeof(*kvs->kvs) * kvs->cap_kvs); + } + kvs->kvs[kvs->num_kvs].key = key; + kvs->kvs[kvs->num_kvs].value = value; + kvs->num_kvs++; +} + +static char *key_int(grpc_error_ints which) { + return gpr_strdup(error_int_name(which)); +} + +static char *fmt_int(intptr_t p) { + char *s; + gpr_asprintf(&s, "%" PRIdPTR, p); + return s; +} + +static void collect_ints_kvs(grpc_error *err, kv_pairs *kvs) { + for (size_t which = 0; which < GRPC_ERROR_INT_MAX; ++which) { + uint8_t slot = err->ints[which]; + if (slot != UINT8_MAX) { + append_kv(kvs, key_int((grpc_error_ints)which), + fmt_int(err->arena[slot])); + } + } +} + +static char *key_str(grpc_error_strs which) { + return gpr_strdup(error_str_name(which)); +} + static char *fmt_str(void *p) { char *s = NULL; size_t sz = 0; @@ -468,8 +595,22 @@ static char *fmt_str(void *p) { return s; } -static char *fmt_time(void *p) { - gpr_timespec tm = *(gpr_timespec *)p; +static void collect_strs_kvs(grpc_error *err, kv_pairs *kvs) { + for (size_t which = 0; which < GRPC_ERROR_STR_MAX; ++which) { + uint8_t slot = err->strs[which]; + if (slot != UINT8_MAX) { + append_kv( + kvs, key_str((grpc_error_strs)which), + fmt_str(GRPC_SLICE_START_PTR(*(grpc_slice *)(err->arena + slot)))); + } + } +} + +static char *key_time(grpc_error_times which) { + return gpr_strdup(error_time_name(which)); +} + +static char *fmt_time(gpr_timespec tm) { char *out; char *pfx = "!!"; switch (tm.clock_type) { @@ -490,24 +631,37 @@ static char *fmt_time(void *p) { return out; } -static void add_errs(gpr_avl_node *n, char **s, size_t *sz, size_t *cap, - bool *first) { - if (n == NULL) return; - add_errs(n->left, s, sz, cap, first); - if (!*first) append_chr(',', s, sz, cap); - *first = false; - const char *e = grpc_error_string(n->value); - append_str(e, s, sz, cap); - add_errs(n->right, s, sz, cap, first); +static void collect_times_kvs(grpc_error *err, kv_pairs *kvs) { + for (size_t which = 0; which < GRPC_ERROR_TIME_MAX; ++which) { + uint8_t slot = err->times[which]; + if (slot != UINT8_MAX) { + append_kv(kvs, key_time((grpc_error_times)which), + fmt_time(*(gpr_timespec *)(err->arena + slot))); + } + } +} + +static void add_errs(grpc_error *err, char **s, size_t *sz, size_t *cap) { + uint8_t slot = err->first_err; + bool first = true; + while (slot != UINT8_MAX) { + grpc_linked_error *lerr = (grpc_linked_error *)(err->arena + slot); + if (!first) append_chr(',', s, sz, cap); + first = false; + const char *e = grpc_error_string(lerr->err); + append_str(e, s, sz, cap); + GPR_ASSERT(err->last_err == slot ? lerr->next == UINT8_MAX + : lerr->next != UINT8_MAX); + slot = lerr->next; + } } static char *errs_string(grpc_error *err) { char *s = NULL; size_t sz = 0; size_t cap = 0; - bool first = true; append_chr('[', &s, &sz, &cap); - add_errs(err->errs.root, &s, &sz, &cap, &first); + add_errs(err, &s, &sz, &cap); append_chr(']', &s, &sz, &cap); append_chr(0, &s, &sz, &cap); return s; @@ -546,7 +700,7 @@ const char *grpc_error_string(grpc_error *err) { if (err == GRPC_ERROR_OOM) return oom_error_string; if (err == GRPC_ERROR_CANCELLED) return cancelled_error_string; - void *p = (void *)gpr_atm_acq_load(&err->error_string); + void *p = (void *)gpr_atm_acq_load(&err->atomics.error_string); if (p != NULL) { GPR_TIMER_END("grpc_error_string", 0); return p; @@ -555,10 +709,10 @@ const char *grpc_error_string(grpc_error *err) { kv_pairs kvs; memset(&kvs, 0, sizeof(kvs)); - collect_kvs(err->ints.root, key_int, fmt_int, &kvs); - collect_kvs(err->strs.root, key_str, fmt_str, &kvs); - collect_kvs(err->times.root, key_time, fmt_time, &kvs); - if (!gpr_avl_is_empty(err->errs)) { + collect_ints_kvs(err, &kvs); + collect_strs_kvs(err, &kvs); + collect_times_kvs(err, &kvs); + if (err->first_err != UINT8_MAX) { append_kv(&kvs, gpr_strdup("referenced_errors"), errs_string(err)); } @@ -566,9 +720,9 @@ const char *grpc_error_string(grpc_error *err) { char *out = finish_kvs(&kvs); - if (!gpr_atm_rel_cas(&err->error_string, 0, (gpr_atm)out)) { + if (!gpr_atm_rel_cas(&err->atomics.error_string, 0, (gpr_atm)out)) { gpr_free(out); - out = (char *)gpr_atm_no_barrier_load(&err->error_string); + out = (char *)gpr_atm_no_barrier_load(&err->atomics.error_string); } GPR_TIMER_END("grpc_error_string", 0); diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h index 2613512acb..eb953947ae 100644 --- a/src/core/lib/iomgr/error.h +++ b/src/core/lib/iomgr/error.h @@ -102,6 +102,9 @@ typedef enum { GRPC_ERROR_INT_LIMIT, /// chttp2: did the error occur while a write was in progress GRPC_ERROR_INT_OCCURRED_DURING_WRITE, + + /// Must always be last + GRPC_ERROR_INT_MAX, } grpc_error_ints; typedef enum { @@ -129,11 +132,17 @@ typedef enum { GRPC_ERROR_STR_KEY, /// value associated with the error GRPC_ERROR_STR_VALUE, + + /// Must always be last + GRPC_ERROR_STR_MAX, } grpc_error_strs; typedef enum { /// timestamp of error creation GRPC_ERROR_TIME_CREATED, + + /// Must always be last + GRPC_ERROR_TIME_MAX, } grpc_error_times; /// The following "special" errors can be propagated without allocating memory. @@ -184,8 +193,6 @@ void grpc_error_unref(grpc_error *err); grpc_error *grpc_error_set_int(grpc_error *src, grpc_error_ints which, intptr_t value) GRPC_MUST_USE_RESULT; bool grpc_error_get_int(grpc_error *error, grpc_error_ints which, intptr_t *p); -grpc_error *grpc_error_set_time(grpc_error *src, grpc_error_times which, - gpr_timespec value) GRPC_MUST_USE_RESULT; grpc_error *grpc_error_set_str(grpc_error *src, grpc_error_strs which, const char *value) GRPC_MUST_USE_RESULT; /// Returns NULL if the specified string is not set. diff --git a/src/core/lib/iomgr/error_internal.h b/src/core/lib/iomgr/error_internal.h index 1c89ead4ed..7f204df1b2 100644 --- a/src/core/lib/iomgr/error_internal.h +++ b/src/core/lib/iomgr/error_internal.h @@ -35,18 +35,39 @@ #define GRPC_CORE_LIB_IOMGR_ERROR_INTERNAL_H #include <inttypes.h> -#include <stdbool.h> +#include <stdbool.h> // TODO, do we need this? -#include <grpc/support/avl.h> +#include <grpc/support/sync.h> +typedef struct grpc_linked_error grpc_linked_error; + +struct grpc_linked_error { + grpc_error *err; + uint8_t next; +}; + +// c core representation of an error. See error.h for high level description of +// this object. struct grpc_error { - gpr_refcount refs; - gpr_avl ints; - gpr_avl strs; - gpr_avl times; - gpr_avl errs; - uintptr_t next_err; - gpr_atm error_string; + // All atomics in grpc_error must be stored in this nested struct. The rest of + // the object is memcpy-ed in bulk in copy_and_unref. + struct atomics { + gpr_refcount refs; + gpr_atm error_string; + } atomics; + // These arrays index into dynamic arena at the bottom of the struct. + // UINT8_MAX is used as a sentinel value. + uint8_t ints[GRPC_ERROR_INT_MAX]; + uint8_t strs[GRPC_ERROR_STR_MAX]; + uint8_t times[GRPC_ERROR_TIME_MAX]; + // The child errors are stored in the arena, but are effectively a linked list + // structure, since they are contained withing grpc_linked_error objects. + uint8_t first_err; + uint8_t last_err; + // The arena is dynamically reallocated with a grow factor of 1.5. + uint8_t arena_size; + uint8_t arena_capacity; + intptr_t arena[0]; }; bool grpc_error_is_special(grpc_error *err); diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c index c03fadaebb..5ddd5313e2 100644 --- a/src/core/lib/iomgr/ev_poll_posix.c +++ b/src/core/lib/iomgr/ev_poll_posix.c @@ -1131,8 +1131,7 @@ static int poll_deadline_to_millis_timeout(gpr_timespec deadline, */ static grpc_pollset_set *pollset_set_create(void) { - grpc_pollset_set *pollset_set = gpr_malloc(sizeof(*pollset_set)); - memset(pollset_set, 0, sizeof(*pollset_set)); + grpc_pollset_set *pollset_set = gpr_zalloc(sizeof(*pollset_set)); gpr_mu_init(&pollset_set->mu); return pollset_set; } diff --git a/src/core/lib/iomgr/pollset.h b/src/core/lib/iomgr/pollset.h index c4b62a2790..e19ce697b8 100644 --- a/src/core/lib/iomgr/pollset.h +++ b/src/core/lib/iomgr/pollset.h @@ -53,6 +53,7 @@ typedef struct grpc_pollset grpc_pollset; typedef struct grpc_pollset_worker grpc_pollset_worker; size_t grpc_pollset_size(void); +/* Initialize a pollset: assumes *pollset contains all zeros */ void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu); /* Begin shutting down the pollset, and call closure when done. * pollset's mutex must be held */ diff --git a/src/core/lib/iomgr/pollset_uv.c b/src/core/lib/iomgr/pollset_uv.c index 5847771108..a2f81bcd78 100644 --- a/src/core/lib/iomgr/pollset_uv.c +++ b/src/core/lib/iomgr/pollset_uv.c @@ -39,6 +39,7 @@ #include <string.h> +#include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/sync.h> @@ -57,24 +58,40 @@ int grpc_pollset_work_run_loop; gpr_mu grpc_polling_mu; +/* This is used solely to kick the uv loop, by setting a callback to be run + immediately in the next loop iteration. + Note: In the future, if there is a bug that involves missing wakeups in the + future, try adding a uv_async_t to kick the loop differently */ +uv_timer_t *dummy_uv_handle; + size_t grpc_pollset_size() { return sizeof(grpc_pollset); } +void dummy_timer_cb(uv_timer_t *handle) {} + +void dummy_handle_close_cb(uv_handle_t *handle) { gpr_free(handle); } + void grpc_pollset_global_init(void) { gpr_mu_init(&grpc_polling_mu); + dummy_uv_handle = gpr_malloc(sizeof(uv_timer_t)); + uv_timer_init(uv_default_loop(), dummy_uv_handle); grpc_pollset_work_run_loop = 1; } -void grpc_pollset_global_shutdown(void) { gpr_mu_destroy(&grpc_polling_mu); } +void grpc_pollset_global_shutdown(void) { + gpr_mu_destroy(&grpc_polling_mu); + uv_close((uv_handle_t *)dummy_uv_handle, dummy_handle_close_cb); +} + +static void timer_run_cb(uv_timer_t *timer) {} + +static void timer_close_cb(uv_handle_t *handle) { handle->data = (void *)1; } void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu) { *mu = &grpc_polling_mu; - memset(pollset, 0, sizeof(grpc_pollset)); uv_timer_init(uv_default_loop(), &pollset->timer); pollset->shutting_down = 0; } -static void timer_close_cb(uv_handle_t *handle) { handle->data = (void *)1; } - void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_closure *closure) { GPR_ASSERT(!pollset->shutting_down); @@ -82,6 +99,9 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, if (grpc_pollset_work_run_loop) { // Drain any pending UV callbacks without blocking uv_run(uv_default_loop(), UV_RUN_NOWAIT); + } else { + // kick the loop once + uv_timer_start(dummy_uv_handle, dummy_timer_cb, 0, 0); } grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE); } @@ -97,8 +117,6 @@ void grpc_pollset_destroy(grpc_pollset *pollset) { } } -static void timer_run_cb(uv_timer_t *timer) {} - grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_pollset_worker **worker_hdl, gpr_timespec now, gpr_timespec deadline) { @@ -131,6 +149,7 @@ grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_error *grpc_pollset_kick(grpc_pollset *pollset, grpc_pollset_worker *specific_worker) { + uv_timer_start(dummy_uv_handle, dummy_timer_cb, 0, 0); return GRPC_ERROR_NONE; } diff --git a/src/core/lib/iomgr/pollset_windows.c b/src/core/lib/iomgr/pollset_windows.c index 9be73a7ce8..17043c1ea1 100644 --- a/src/core/lib/iomgr/pollset_windows.c +++ b/src/core/lib/iomgr/pollset_windows.c @@ -98,7 +98,6 @@ size_t grpc_pollset_size(void) { return sizeof(grpc_pollset); } void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu) { *mu = &grpc_polling_mu; - memset(pollset, 0, sizeof(*pollset)); pollset->root_worker.links[GRPC_POLLSET_WORKER_LINK_POLLSET].next = pollset->root_worker.links[GRPC_POLLSET_WORKER_LINK_POLLSET].prev = &pollset->root_worker; diff --git a/src/core/lib/iomgr/port.h b/src/core/lib/iomgr/port.h index f1897bb91f..94a454c0b7 100644 --- a/src/core/lib/iomgr/port.h +++ b/src/core/lib/iomgr/port.h @@ -39,6 +39,7 @@ #if defined(GRPC_UV) // Do nothing #elif defined(GPR_MANYLINUX1) +#define GRPC_HAVE_IFADDRS 1 #define GRPC_HAVE_IPV6_RECVPKTINFO 1 #define GRPC_HAVE_IP_PKTINFO 1 #define GRPC_HAVE_MSG_NOSIGNAL 1 @@ -65,6 +66,7 @@ #define GRPC_POSIX_WAKEUP_FD 1 #define GRPC_TIMER_USE_GENERIC 1 #elif defined(GPR_LINUX) +#define GRPC_HAVE_IFADDRS 1 #define GRPC_HAVE_IPV6_RECVPKTINFO 1 #define GRPC_HAVE_IP_PKTINFO 1 #define GRPC_HAVE_MSG_NOSIGNAL 1 @@ -90,6 +92,7 @@ #define GRPC_POSIX_SOCKETUTILS #endif #elif defined(GPR_APPLE) +#define GRPC_HAVE_IFADDRS 1 #define GRPC_HAVE_SO_NOSIGPIPE 1 #define GRPC_HAVE_UNIX_SOCKET 1 #define GRPC_MSG_IOVLEN_TYPE int @@ -100,6 +103,7 @@ #define GRPC_POSIX_WAKEUP_FD 1 #define GRPC_TIMER_USE_GENERIC 1 #elif defined(GPR_FREEBSD) +#define GRPC_HAVE_IFADDRS 1 #define GRPC_HAVE_IPV6_RECVPKTINFO 1 #define GRPC_HAVE_SO_NOSIGPIPE 1 #define GRPC_HAVE_UNIX_SOCKET 1 diff --git a/src/core/lib/iomgr/resolve_address_uv.c b/src/core/lib/iomgr/resolve_address_uv.c index 79ff910738..4d715be94c 100644 --- a/src/core/lib/iomgr/resolve_address_uv.c +++ b/src/core/lib/iomgr/resolve_address_uv.c @@ -40,6 +40,7 @@ #include <grpc/support/host_port.h> #include <grpc/support/log.h> #include <grpc/support/string_util.h> +#include <grpc/support/useful.h> #include "src/core/lib/iomgr/closure.h" #include "src/core/lib/iomgr/error.h" @@ -54,8 +55,36 @@ typedef struct request { grpc_closure *on_done; grpc_resolved_addresses **addresses; struct addrinfo *hints; + char *host; + char *port; } request; +static int retry_named_port_failure(int status, request *r, + uv_getaddrinfo_cb getaddrinfo_cb) { + if (status != 0) { + // This loop is copied from resolve_address_posix.c + char *svc[][2] = {{"http", "80"}, {"https", "443"}}; + for (size_t i = 0; i < GPR_ARRAY_SIZE(svc); i++) { + if (strcmp(r->port, svc[i][0]) == 0) { + int retry_status; + uv_getaddrinfo_t *req = gpr_malloc(sizeof(uv_getaddrinfo_t)); + req->data = r; + retry_status = uv_getaddrinfo(uv_default_loop(), req, getaddrinfo_cb, + r->host, svc[i][1], r->hints); + if (retry_status < 0 || getaddrinfo_cb == NULL) { + // The callback will not be called + gpr_free(req); + } + return retry_status; + } + } + } + /* If this function calls uv_getaddrinfo, it will return that function's + return value. That function only returns numbers <=0, so we can safely + return 1 to indicate that we never retried */ + return 1; +} + static grpc_error *handle_addrinfo_result(int status, struct addrinfo *result, grpc_resolved_addresses **addresses) { struct addrinfo *resp; @@ -97,13 +126,21 @@ static void getaddrinfo_callback(uv_getaddrinfo_t *req, int status, request *r = (request *)req->data; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_error *error; + int retry_status; + + gpr_free(req); + retry_status = retry_named_port_failure(status, r, getaddrinfo_callback); + if (retry_status == 0) { + // The request is being retried. Nothing should be done here + return; + } + /* Either no retry was attempted, or the retry failed. Either way, the + original error probably has more interesting information */ error = handle_addrinfo_result(status, res, r->addresses); grpc_closure_sched(&exec_ctx, r->on_done, error); grpc_exec_ctx_finish(&exec_ctx); - gpr_free(r->hints); gpr_free(r); - gpr_free(req); uv_freeaddrinfo(res); } @@ -143,6 +180,7 @@ static grpc_error *blocking_resolve_address_impl( uv_getaddrinfo_t req; int s; grpc_error *err; + int retry_status; req.addrinfo = NULL; @@ -158,6 +196,12 @@ static grpc_error *blocking_resolve_address_impl( hints.ai_flags = AI_PASSIVE; /* for wildcard IP address */ s = uv_getaddrinfo(uv_default_loop(), &req, NULL, host, port, &hints); + request r = { + .addresses = addresses, .hints = &hints, .host = host, .port = port}; + retry_status = retry_named_port_failure(s, &r, NULL); + if (retry_status <= 0) { + s = retry_status; + } err = handle_addrinfo_result(s, req.addrinfo, addresses); done: @@ -200,6 +244,8 @@ static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name, r = gpr_malloc(sizeof(request)); r->on_done = on_done; r->addresses = addrs; + r->host = host; + r->port = port; req = gpr_malloc(sizeof(uv_getaddrinfo_t)); req->data = r; @@ -222,6 +268,8 @@ static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name, gpr_free(r); gpr_free(req); gpr_free(hints); + gpr_free(host); + gpr_free(port); } } diff --git a/src/core/lib/iomgr/sockaddr_utils.c b/src/core/lib/iomgr/sockaddr_utils.c index ffa62cb53c..9d2732666b 100644 --- a/src/core/lib/iomgr/sockaddr_utils.c +++ b/src/core/lib/iomgr/sockaddr_utils.c @@ -162,6 +162,7 @@ int grpc_sockaddr_to_string(char **out, char ntop_buf[INET6_ADDRSTRLEN]; const void *ip = NULL; int port; + uint32_t sin6_scope_id = 0; int ret; *out = NULL; @@ -177,10 +178,19 @@ int grpc_sockaddr_to_string(char **out, const struct sockaddr_in6 *addr6 = (const struct sockaddr_in6 *)addr; ip = &addr6->sin6_addr; port = ntohs(addr6->sin6_port); + sin6_scope_id = addr6->sin6_scope_id; } if (ip != NULL && grpc_inet_ntop(addr->sa_family, ip, ntop_buf, sizeof(ntop_buf)) != NULL) { - ret = gpr_join_host_port(out, ntop_buf, port); + if (sin6_scope_id != 0) { + char *host_with_scope; + /* Enclose sin6_scope_id with the format defined in RFC 6784 section 2. */ + gpr_asprintf(&host_with_scope, "%s%%25%" PRIu32, ntop_buf, sin6_scope_id); + ret = gpr_join_host_port(out, host_with_scope, port); + gpr_free(host_with_scope); + } else { + ret = gpr_join_host_port(out, ntop_buf, port); + } } else { ret = gpr_asprintf(out, "(sockaddr family=%d)", addr->sa_family); } diff --git a/src/core/lib/iomgr/tcp_client_uv.c b/src/core/lib/iomgr/tcp_client_uv.c index 3de0795187..618483d9cb 100644 --- a/src/core/lib/iomgr/tcp_client_uv.c +++ b/src/core/lib/iomgr/tcp_client_uv.c @@ -76,7 +76,6 @@ static void uv_tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, const char *str = grpc_error_string(error); gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_alarm: error=%s", connect->addr_name, str); - grpc_error_free_string(str); } if (error == GRPC_ERROR_NONE) { /* error == NONE implies that the timer ran out, and wasn't cancelled. If @@ -144,8 +143,7 @@ static void tcp_client_connect_impl(grpc_exec_ctx *exec_ctx, } } - connect = gpr_malloc(sizeof(grpc_uv_tcp_connect)); - memset(connect, 0, sizeof(grpc_uv_tcp_connect)); + connect = gpr_zalloc(sizeof(grpc_uv_tcp_connect)); connect->closure = closure; connect->endpoint = ep; connect->tcp_handle = gpr_malloc(sizeof(uv_tcp_t)); diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c index 36f878fdd4..b3644518f5 100644 --- a/src/core/lib/iomgr/tcp_server_posix.c +++ b/src/core/lib/iomgr/tcp_server_posix.c @@ -44,11 +44,8 @@ #include <errno.h> #include <fcntl.h> -#include <ifaddrs.h> -#include <limits.h> #include <netinet/in.h> #include <netinet/tcp.h> -#include <stdio.h> #include <string.h> #include <sys/socket.h> #include <sys/stat.h> @@ -67,80 +64,10 @@ #include "src/core/lib/iomgr/sockaddr_utils.h" #include "src/core/lib/iomgr/socket_utils_posix.h" #include "src/core/lib/iomgr/tcp_posix.h" +#include "src/core/lib/iomgr/tcp_server_utils_posix.h" #include "src/core/lib/iomgr/unix_sockets_posix.h" #include "src/core/lib/support/string.h" -#define MIN_SAFE_ACCEPT_QUEUE_SIZE 100 - -static gpr_once s_init_max_accept_queue_size; -static int s_max_accept_queue_size; - -/* one listening port */ -typedef struct grpc_tcp_listener grpc_tcp_listener; -struct grpc_tcp_listener { - int fd; - grpc_fd *emfd; - grpc_tcp_server *server; - grpc_resolved_address addr; - int port; - unsigned port_index; - unsigned fd_index; - grpc_closure read_closure; - grpc_closure destroyed_closure; - struct grpc_tcp_listener *next; - /* sibling is a linked list of all listeners for a given port. add_port and - clone_port place all new listeners in the same sibling list. A member of - the 'sibling' list is also a member of the 'next' list. The head of each - sibling list has is_sibling==0, and subsequent members of sibling lists - have is_sibling==1. is_sibling allows separate sibling lists to be - identified while iterating through 'next'. */ - struct grpc_tcp_listener *sibling; - int is_sibling; -}; - -/* the overall server */ -struct grpc_tcp_server { - gpr_refcount refs; - /* Called whenever accept() succeeds on a server port. */ - grpc_tcp_server_cb on_accept_cb; - void *on_accept_cb_arg; - - gpr_mu mu; - - /* active port count: how many ports are actually still listening */ - size_t active_ports; - /* destroyed port count: how many ports are completely destroyed */ - size_t destroyed_ports; - - /* is this server shutting down? */ - bool shutdown; - /* use SO_REUSEPORT */ - bool so_reuseport; - /* expand wildcard addresses to a list of all local addresses */ - bool expand_wildcard_addrs; - - /* linked list of server ports */ - grpc_tcp_listener *head; - grpc_tcp_listener *tail; - unsigned nports; - - /* List of closures passed to shutdown_starting_add(). */ - grpc_closure_list shutdown_starting; - - /* shutdown callback */ - grpc_closure *shutdown_complete; - - /* all pollsets interested in new connections */ - grpc_pollset **pollsets; - /* number of pollsets in the pollsets array */ - size_t pollset_count; - - /* next pollset to assign a channel to */ - gpr_atm next_pollset_to_assign; - - grpc_resource_quota *resource_quota; -}; - static gpr_once check_init = GPR_ONCE_INIT; static bool has_so_reuseport = false; @@ -161,7 +88,7 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx, grpc_tcp_server **server) { gpr_once_init(&check_init, init); - grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server)); + grpc_tcp_server *s = gpr_zalloc(sizeof(grpc_tcp_server)); s->so_reuseport = has_so_reuseport; s->resource_quota = grpc_resource_quota_create(NULL); s->expand_wildcard_addrs = false; @@ -299,99 +226,6 @@ static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { } } -/* get max listen queue size on linux */ -static void init_max_accept_queue_size(void) { - int n = SOMAXCONN; - char buf[64]; - FILE *fp = fopen("/proc/sys/net/core/somaxconn", "r"); - if (fp == NULL) { - /* 2.4 kernel. */ - s_max_accept_queue_size = SOMAXCONN; - return; - } - if (fgets(buf, sizeof buf, fp)) { - char *end; - long i = strtol(buf, &end, 10); - if (i > 0 && i <= INT_MAX && end && *end == 0) { - n = (int)i; - } - } - fclose(fp); - s_max_accept_queue_size = n; - - if (s_max_accept_queue_size < MIN_SAFE_ACCEPT_QUEUE_SIZE) { - gpr_log(GPR_INFO, - "Suspiciously small accept queue (%d) will probably lead to " - "connection drops", - s_max_accept_queue_size); - } -} - -static int get_max_accept_queue_size(void) { - gpr_once_init(&s_init_max_accept_queue_size, init_max_accept_queue_size); - return s_max_accept_queue_size; -} - -/* Prepare a recently-created socket for listening. */ -static grpc_error *prepare_socket(int fd, const grpc_resolved_address *addr, - bool so_reuseport, int *port) { - grpc_resolved_address sockname_temp; - grpc_error *err = GRPC_ERROR_NONE; - - GPR_ASSERT(fd >= 0); - - if (so_reuseport && !grpc_is_unix_socket(addr)) { - err = grpc_set_socket_reuse_port(fd, 1); - if (err != GRPC_ERROR_NONE) goto error; - } - - err = grpc_set_socket_nonblocking(fd, 1); - if (err != GRPC_ERROR_NONE) goto error; - err = grpc_set_socket_cloexec(fd, 1); - if (err != GRPC_ERROR_NONE) goto error; - if (!grpc_is_unix_socket(addr)) { - err = grpc_set_socket_low_latency(fd, 1); - if (err != GRPC_ERROR_NONE) goto error; - err = grpc_set_socket_reuse_addr(fd, 1); - if (err != GRPC_ERROR_NONE) goto error; - } - err = grpc_set_socket_no_sigpipe_if_possible(fd); - if (err != GRPC_ERROR_NONE) goto error; - - GPR_ASSERT(addr->len < ~(socklen_t)0); - if (bind(fd, (struct sockaddr *)addr->addr, (socklen_t)addr->len) < 0) { - err = GRPC_OS_ERROR(errno, "bind"); - goto error; - } - - if (listen(fd, get_max_accept_queue_size()) < 0) { - err = GRPC_OS_ERROR(errno, "listen"); - goto error; - } - - sockname_temp.len = sizeof(struct sockaddr_storage); - - if (getsockname(fd, (struct sockaddr *)sockname_temp.addr, - (socklen_t *)&sockname_temp.len) < 0) { - err = GRPC_OS_ERROR(errno, "getsockname"); - goto error; - } - - *port = grpc_sockaddr_get_port(&sockname_temp); - return GRPC_ERROR_NONE; - -error: - GPR_ASSERT(err != GRPC_ERROR_NONE); - if (fd >= 0) { - close(fd); - } - grpc_error *ret = grpc_error_set_int( - GRPC_ERROR_CREATE_REFERENCING("Unable to configure socket", &err, 1), - GRPC_ERROR_INT_FD, fd); - GRPC_ERROR_UNREF(err); - return ret; -} - /* event manager callback when reads are ready */ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *err) { grpc_tcp_listener *sp = arg; @@ -422,7 +256,14 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *err) { grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure); return; default: - gpr_log(GPR_ERROR, "Failed accept4: %s", strerror(errno)); + gpr_mu_lock(&sp->server->mu); + if (!sp->server->shutdown_listeners) { + gpr_log(GPR_ERROR, "Failed accept4: %s", strerror(errno)); + } else { + /* if we have shutdown listeners, accept4 could fail, and we + needn't notify users */ + } + gpr_mu_unlock(&sp->server->mu); goto error; } } @@ -438,11 +279,6 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *err) { grpc_fd *fdobj = grpc_fd_create(fd, name); - if (read_notifier_pollset == NULL) { - gpr_log(GPR_ERROR, "Read notifier pollset is not set on the fd"); - goto error; - } - grpc_pollset_add_fd(exec_ctx, read_notifier_pollset, fdobj); // Create acceptor. @@ -473,216 +309,6 @@ error: } } -static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd, - const grpc_resolved_address *addr, - unsigned port_index, unsigned fd_index, - grpc_tcp_listener **listener) { - grpc_tcp_listener *sp = NULL; - int port = -1; - char *addr_str; - char *name; - - grpc_error *err = prepare_socket(fd, addr, s->so_reuseport, &port); - if (err == GRPC_ERROR_NONE) { - GPR_ASSERT(port > 0); - grpc_sockaddr_to_string(&addr_str, addr, 1); - gpr_asprintf(&name, "tcp-server-listener:%s", addr_str); - gpr_mu_lock(&s->mu); - s->nports++; - GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server"); - sp = gpr_malloc(sizeof(grpc_tcp_listener)); - sp->next = NULL; - if (s->head == NULL) { - s->head = sp; - } else { - s->tail->next = sp; - } - s->tail = sp; - sp->server = s; - sp->fd = fd; - sp->emfd = grpc_fd_create(fd, name); - memcpy(&sp->addr, addr, sizeof(grpc_resolved_address)); - sp->port = port; - sp->port_index = port_index; - sp->fd_index = fd_index; - sp->is_sibling = 0; - sp->sibling = NULL; - GPR_ASSERT(sp->emfd); - gpr_mu_unlock(&s->mu); - gpr_free(addr_str); - gpr_free(name); - } - - *listener = sp; - return err; -} - -/* If successful, add a listener to s for addr, set *dsmode for the socket, and - return the *listener. */ -static grpc_error *add_addr_to_server(grpc_tcp_server *s, - const grpc_resolved_address *addr, - unsigned port_index, unsigned fd_index, - grpc_dualstack_mode *dsmode, - grpc_tcp_listener **listener) { - grpc_resolved_address addr4_copy; - int fd; - grpc_error *err = - grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, dsmode, &fd); - if (err != GRPC_ERROR_NONE) { - return err; - } - if (*dsmode == GRPC_DSMODE_IPV4 && - grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) { - addr = &addr4_copy; - } - return add_socket_to_server(s, fd, addr, port_index, fd_index, listener); -} - -/* Bind to "::" to get a port number not used by any address. */ -static grpc_error *get_unused_port(int *port) { - grpc_resolved_address wild; - grpc_sockaddr_make_wildcard6(0, &wild); - grpc_dualstack_mode dsmode; - int fd; - grpc_error *err = - grpc_create_dualstack_socket(&wild, SOCK_STREAM, 0, &dsmode, &fd); - if (err != GRPC_ERROR_NONE) { - return err; - } - if (dsmode == GRPC_DSMODE_IPV4) { - grpc_sockaddr_make_wildcard4(0, &wild); - } - if (bind(fd, (const struct sockaddr *)wild.addr, (socklen_t)wild.len) != 0) { - err = GRPC_OS_ERROR(errno, "bind"); - close(fd); - return err; - } - if (getsockname(fd, (struct sockaddr *)wild.addr, (socklen_t *)&wild.len) != - 0) { - err = GRPC_OS_ERROR(errno, "getsockname"); - close(fd); - return err; - } - close(fd); - *port = grpc_sockaddr_get_port(&wild); - return *port <= 0 ? GRPC_ERROR_CREATE("Bad port") : GRPC_ERROR_NONE; -} - -/* Return the listener in s with address addr or NULL. */ -static grpc_tcp_listener *find_listener_with_addr(grpc_tcp_server *s, - grpc_resolved_address *addr) { - grpc_tcp_listener *l; - gpr_mu_lock(&s->mu); - for (l = s->head; l != NULL; l = l->next) { - if (l->addr.len != addr->len) { - continue; - } - if (memcmp(l->addr.addr, addr->addr, addr->len) == 0) { - break; - } - } - gpr_mu_unlock(&s->mu); - return l; -} - -/* Get all addresses assigned to network interfaces on the machine and create a - listener for each. requested_port is the port to use for every listener, or 0 - to select one random port that will be used for every listener. Set *out_port - to the port selected. Return GRPC_ERROR_NONE only if all listeners were - added. */ -static grpc_error *add_all_local_addrs_to_server(grpc_tcp_server *s, - unsigned port_index, - int requested_port, - int *out_port) { - struct ifaddrs *ifa = NULL; - struct ifaddrs *ifa_it; - unsigned fd_index = 0; - grpc_tcp_listener *sp = NULL; - grpc_error *err = GRPC_ERROR_NONE; - if (requested_port == 0) { - /* Note: There could be a race where some local addrs can listen on the - selected port and some can't. The sane way to handle this would be to - retry by recreating the whole grpc_tcp_server. Backing out individual - listeners and orphaning the FDs looks like too much trouble. */ - if ((err = get_unused_port(&requested_port)) != GRPC_ERROR_NONE) { - return err; - } else if (requested_port <= 0) { - return GRPC_ERROR_CREATE("Bad get_unused_port()"); - } - gpr_log(GPR_DEBUG, "Picked unused port %d", requested_port); - } - if (getifaddrs(&ifa) != 0 || ifa == NULL) { - return GRPC_OS_ERROR(errno, "getifaddrs"); - } - for (ifa_it = ifa; ifa_it != NULL; ifa_it = ifa_it->ifa_next) { - grpc_resolved_address addr; - char *addr_str = NULL; - grpc_dualstack_mode dsmode; - grpc_tcp_listener *new_sp = NULL; - const char *ifa_name = (ifa_it->ifa_name ? ifa_it->ifa_name : "<unknown>"); - if (ifa_it->ifa_addr == NULL) { - continue; - } else if (ifa_it->ifa_addr->sa_family == AF_INET) { - addr.len = sizeof(struct sockaddr_in); - } else if (ifa_it->ifa_addr->sa_family == AF_INET6) { - addr.len = sizeof(struct sockaddr_in6); - } else { - continue; - } - memcpy(addr.addr, ifa_it->ifa_addr, addr.len); - if (!grpc_sockaddr_set_port(&addr, requested_port)) { - /* Should never happen, because we check sa_family above. */ - err = GRPC_ERROR_CREATE("Failed to set port"); - break; - } - if (grpc_sockaddr_to_string(&addr_str, &addr, 0) < 0) { - addr_str = gpr_strdup("<error>"); - } - gpr_log(GPR_DEBUG, - "Adding local addr from interface %s flags 0x%x to server: %s", - ifa_name, ifa_it->ifa_flags, addr_str); - /* We could have multiple interfaces with the same address (e.g., bonding), - so look for duplicates. */ - if (find_listener_with_addr(s, &addr) != NULL) { - gpr_log(GPR_DEBUG, "Skipping duplicate addr %s on interface %s", addr_str, - ifa_name); - gpr_free(addr_str); - continue; - } - if ((err = add_addr_to_server(s, &addr, port_index, fd_index, &dsmode, - &new_sp)) != GRPC_ERROR_NONE) { - char *err_str = NULL; - grpc_error *root_err; - if (gpr_asprintf(&err_str, "Failed to add listener: %s", addr_str) < 0) { - err_str = gpr_strdup("Failed to add listener"); - } - root_err = GRPC_ERROR_CREATE(err_str); - gpr_free(err_str); - gpr_free(addr_str); - err = grpc_error_add_child(root_err, err); - break; - } else { - GPR_ASSERT(requested_port == new_sp->port); - ++fd_index; - if (sp != NULL) { - new_sp->is_sibling = 1; - sp->sibling = new_sp; - } - sp = new_sp; - } - gpr_free(addr_str); - } - freeifaddrs(ifa); - if (err != GRPC_ERROR_NONE) { - return err; - } else if (sp == NULL) { - return GRPC_ERROR_CREATE("No local addresses"); - } else { - *out_port = sp->port; - return GRPC_ERROR_NONE; - } -} - /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */ static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s, unsigned port_index, @@ -697,14 +323,16 @@ static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s, grpc_error *v6_err = GRPC_ERROR_NONE; grpc_error *v4_err = GRPC_ERROR_NONE; *out_port = -1; - if (s->expand_wildcard_addrs) { - return add_all_local_addrs_to_server(s, port_index, requested_port, - out_port); + + if (grpc_tcp_server_have_ifaddrs() && s->expand_wildcard_addrs) { + return grpc_tcp_server_add_all_local_addrs(s, port_index, requested_port, + out_port); } + grpc_sockaddr_make_wildcards(requested_port, &wild4, &wild6); /* Try listening on IPv6 first. */ - if ((v6_err = add_addr_to_server(s, &wild6, port_index, fd_index, &dsmode, - &sp)) == GRPC_ERROR_NONE) { + if ((v6_err = grpc_tcp_server_add_addr(s, &wild6, port_index, fd_index, + &dsmode, &sp)) == GRPC_ERROR_NONE) { ++fd_index; requested_port = *out_port = sp->port; if (dsmode == GRPC_DSMODE_DUALSTACK || dsmode == GRPC_DSMODE_IPV4) { @@ -713,8 +341,8 @@ static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s, } /* If we got a v6-only socket or nothing, try adding 0.0.0.0. */ grpc_sockaddr_set_port(&wild4, requested_port); - if ((v4_err = add_addr_to_server(s, &wild4, port_index, fd_index, &dsmode, - &sp2)) == GRPC_ERROR_NONE) { + if ((v4_err = grpc_tcp_server_add_addr(s, &wild4, port_index, fd_index, + &dsmode, &sp2)) == GRPC_ERROR_NONE) { *out_port = sp2->port; if (sp != NULL) { sp2->is_sibling = 1; @@ -752,7 +380,7 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) { err = grpc_create_dualstack_socket(&listener->addr, SOCK_STREAM, 0, &dsmode, &fd); if (err != GRPC_ERROR_NONE) return err; - err = prepare_socket(fd, &listener->addr, true, &port); + err = grpc_tcp_server_prepare_socket(fd, &listener->addr, true, &port); if (err != GRPC_ERROR_NONE) return err; listener->server->nports++; grpc_sockaddr_to_string(&addr_str, &listener->addr, 1); @@ -824,7 +452,7 @@ grpc_error *grpc_tcp_server_add_port(grpc_tcp_server *s, if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) { addr = &addr6_v4mapped; } - if ((err = add_addr_to_server(s, addr, port_index, 0, &dsmode, &sp)) == + if ((err = grpc_tcp_server_add_addr(s, addr, port_index, 0, &dsmode, &sp)) == GRPC_ERROR_NONE) { *out_port = sp->port; } @@ -941,6 +569,7 @@ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { void grpc_tcp_server_shutdown_listeners(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { gpr_mu_lock(&s->mu); + s->shutdown_listeners = true; /* shutdown all fd's */ if (s->active_ports) { grpc_tcp_listener *sp; diff --git a/src/core/lib/iomgr/tcp_server_utils_posix.h b/src/core/lib/iomgr/tcp_server_utils_posix.h new file mode 100644 index 0000000000..f5dc8532f9 --- /dev/null +++ b/src/core/lib/iomgr/tcp_server_utils_posix.h @@ -0,0 +1,134 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_IOMGR_TCP_SERVER_UTILS_POSIX_H +#define GRPC_CORE_LIB_IOMGR_TCP_SERVER_UTILS_POSIX_H + +#include "src/core/lib/iomgr/ev_posix.h" +#include "src/core/lib/iomgr/resolve_address.h" +#include "src/core/lib/iomgr/socket_utils_posix.h" +#include "src/core/lib/iomgr/tcp_server.h" + +/* one listening port */ +typedef struct grpc_tcp_listener { + int fd; + grpc_fd *emfd; + grpc_tcp_server *server; + grpc_resolved_address addr; + int port; + unsigned port_index; + unsigned fd_index; + grpc_closure read_closure; + grpc_closure destroyed_closure; + struct grpc_tcp_listener *next; + /* sibling is a linked list of all listeners for a given port. add_port and + clone_port place all new listeners in the same sibling list. A member of + the 'sibling' list is also a member of the 'next' list. The head of each + sibling list has is_sibling==0, and subsequent members of sibling lists + have is_sibling==1. is_sibling allows separate sibling lists to be + identified while iterating through 'next'. */ + struct grpc_tcp_listener *sibling; + int is_sibling; +} grpc_tcp_listener; + +/* the overall server */ +struct grpc_tcp_server { + gpr_refcount refs; + /* Called whenever accept() succeeds on a server port. */ + grpc_tcp_server_cb on_accept_cb; + void *on_accept_cb_arg; + + gpr_mu mu; + + /* active port count: how many ports are actually still listening */ + size_t active_ports; + /* destroyed port count: how many ports are completely destroyed */ + size_t destroyed_ports; + + /* is this server shutting down? */ + bool shutdown; + /* have listeners been shutdown? */ + bool shutdown_listeners; + /* use SO_REUSEPORT */ + bool so_reuseport; + /* expand wildcard addresses to a list of all local addresses */ + bool expand_wildcard_addrs; + + /* linked list of server ports */ + grpc_tcp_listener *head; + grpc_tcp_listener *tail; + unsigned nports; + + /* List of closures passed to shutdown_starting_add(). */ + grpc_closure_list shutdown_starting; + + /* shutdown callback */ + grpc_closure *shutdown_complete; + + /* all pollsets interested in new connections */ + grpc_pollset **pollsets; + /* number of pollsets in the pollsets array */ + size_t pollset_count; + + /* next pollset to assign a channel to */ + gpr_atm next_pollset_to_assign; + + grpc_resource_quota *resource_quota; +}; + +/* If successful, add a listener to \a s for \a addr, set \a dsmode for the + socket, and return the \a listener. */ +grpc_error *grpc_tcp_server_add_addr(grpc_tcp_server *s, + const grpc_resolved_address *addr, + unsigned port_index, unsigned fd_index, + grpc_dualstack_mode *dsmode, + grpc_tcp_listener **listener); + +/* Get all addresses assigned to network interfaces on the machine and create a + listener for each. requested_port is the port to use for every listener, or 0 + to select one random port that will be used for every listener. Set *out_port + to the port selected. Return GRPC_ERROR_NONE only if all listeners were + added. */ +grpc_error *grpc_tcp_server_add_all_local_addrs(grpc_tcp_server *s, + unsigned port_index, + int requested_port, + int *out_port); + +/* Prepare a recently-created socket for listening. */ +grpc_error *grpc_tcp_server_prepare_socket(int fd, + const grpc_resolved_address *addr, + bool so_reuseport, int *port); +/* Ruturn true if the platform supports ifaddrs */ +bool grpc_tcp_server_have_ifaddrs(void); + +#endif /* GRPC_CORE_LIB_IOMGR_TCP_SERVER_UTILS_POSIX_H */ diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_common.c b/src/core/lib/iomgr/tcp_server_utils_posix_common.c new file mode 100644 index 0000000000..e45e27d5ab --- /dev/null +++ b/src/core/lib/iomgr/tcp_server_utils_posix_common.c @@ -0,0 +1,220 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/iomgr/port.h" + +#ifdef GRPC_HAVE_IFADDRS + +#include "src/core/lib/iomgr/tcp_server_utils_posix.h" + +#include <errno.h> +#include <limits.h> +#include <stdio.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <grpc/support/sync.h> + +#include "src/core/lib/iomgr/error.h" +#include "src/core/lib/iomgr/sockaddr.h" +#include "src/core/lib/iomgr/sockaddr_utils.h" +#include "src/core/lib/iomgr/unix_sockets_posix.h" + +#define MIN_SAFE_ACCEPT_QUEUE_SIZE 100 + +static gpr_once s_init_max_accept_queue_size; +static int s_max_accept_queue_size; + +/* get max listen queue size on linux */ +static void init_max_accept_queue_size(void) { + int n = SOMAXCONN; + char buf[64]; + FILE *fp = fopen("/proc/sys/net/core/somaxconn", "r"); + if (fp == NULL) { + /* 2.4 kernel. */ + s_max_accept_queue_size = SOMAXCONN; + return; + } + if (fgets(buf, sizeof buf, fp)) { + char *end; + long i = strtol(buf, &end, 10); + if (i > 0 && i <= INT_MAX && end && *end == 0) { + n = (int)i; + } + } + fclose(fp); + s_max_accept_queue_size = n; + + if (s_max_accept_queue_size < MIN_SAFE_ACCEPT_QUEUE_SIZE) { + gpr_log(GPR_INFO, + "Suspiciously small accept queue (%d) will probably lead to " + "connection drops", + s_max_accept_queue_size); + } +} + +static int get_max_accept_queue_size(void) { + gpr_once_init(&s_init_max_accept_queue_size, init_max_accept_queue_size); + return s_max_accept_queue_size; +} + +static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd, + const grpc_resolved_address *addr, + unsigned port_index, unsigned fd_index, + grpc_tcp_listener **listener) { + grpc_tcp_listener *sp = NULL; + int port = -1; + char *addr_str; + char *name; + + grpc_error *err = + grpc_tcp_server_prepare_socket(fd, addr, s->so_reuseport, &port); + if (err == GRPC_ERROR_NONE) { + GPR_ASSERT(port > 0); + grpc_sockaddr_to_string(&addr_str, addr, 1); + gpr_asprintf(&name, "tcp-server-listener:%s", addr_str); + gpr_mu_lock(&s->mu); + s->nports++; + GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server"); + sp = gpr_malloc(sizeof(grpc_tcp_listener)); + sp->next = NULL; + if (s->head == NULL) { + s->head = sp; + } else { + s->tail->next = sp; + } + s->tail = sp; + sp->server = s; + sp->fd = fd; + sp->emfd = grpc_fd_create(fd, name); + memcpy(&sp->addr, addr, sizeof(grpc_resolved_address)); + sp->port = port; + sp->port_index = port_index; + sp->fd_index = fd_index; + sp->is_sibling = 0; + sp->sibling = NULL; + GPR_ASSERT(sp->emfd); + gpr_mu_unlock(&s->mu); + gpr_free(addr_str); + gpr_free(name); + } + + *listener = sp; + return err; +} + +/* If successful, add a listener to s for addr, set *dsmode for the socket, and + return the *listener. */ +grpc_error *grpc_tcp_server_add_addr(grpc_tcp_server *s, + const grpc_resolved_address *addr, + unsigned port_index, unsigned fd_index, + grpc_dualstack_mode *dsmode, + grpc_tcp_listener **listener) { + grpc_resolved_address addr4_copy; + int fd; + grpc_error *err = + grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, dsmode, &fd); + if (err != GRPC_ERROR_NONE) { + return err; + } + if (*dsmode == GRPC_DSMODE_IPV4 && + grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) { + addr = &addr4_copy; + } + return add_socket_to_server(s, fd, addr, port_index, fd_index, listener); +} + +/* Prepare a recently-created socket for listening. */ +grpc_error *grpc_tcp_server_prepare_socket(int fd, + const grpc_resolved_address *addr, + bool so_reuseport, int *port) { + grpc_resolved_address sockname_temp; + grpc_error *err = GRPC_ERROR_NONE; + + GPR_ASSERT(fd >= 0); + + if (so_reuseport && !grpc_is_unix_socket(addr)) { + err = grpc_set_socket_reuse_port(fd, 1); + if (err != GRPC_ERROR_NONE) goto error; + } + + err = grpc_set_socket_nonblocking(fd, 1); + if (err != GRPC_ERROR_NONE) goto error; + err = grpc_set_socket_cloexec(fd, 1); + if (err != GRPC_ERROR_NONE) goto error; + if (!grpc_is_unix_socket(addr)) { + err = grpc_set_socket_low_latency(fd, 1); + if (err != GRPC_ERROR_NONE) goto error; + err = grpc_set_socket_reuse_addr(fd, 1); + if (err != GRPC_ERROR_NONE) goto error; + } + err = grpc_set_socket_no_sigpipe_if_possible(fd); + if (err != GRPC_ERROR_NONE) goto error; + + GPR_ASSERT(addr->len < ~(socklen_t)0); + if (bind(fd, (struct sockaddr *)addr->addr, (socklen_t)addr->len) < 0) { + err = GRPC_OS_ERROR(errno, "bind"); + goto error; + } + + if (listen(fd, get_max_accept_queue_size()) < 0) { + err = GRPC_OS_ERROR(errno, "listen"); + goto error; + } + + sockname_temp.len = sizeof(struct sockaddr_storage); + + if (getsockname(fd, (struct sockaddr *)sockname_temp.addr, + (socklen_t *)&sockname_temp.len) < 0) { + err = GRPC_OS_ERROR(errno, "getsockname"); + goto error; + } + + *port = grpc_sockaddr_get_port(&sockname_temp); + return GRPC_ERROR_NONE; + +error: + GPR_ASSERT(err != GRPC_ERROR_NONE); + if (fd >= 0) { + close(fd); + } + grpc_error *ret = grpc_error_set_int( + GRPC_ERROR_CREATE_REFERENCING("Unable to configure socket", &err, 1), + GRPC_ERROR_INT_FD, fd); + GRPC_ERROR_UNREF(err); + return ret; +} + +#endif /* GRPC_HAVE_IFADDRS */ diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.c b/src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.c new file mode 100644 index 0000000000..6354a6bdc1 --- /dev/null +++ b/src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.c @@ -0,0 +1,195 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/iomgr/port.h" + +#ifdef GRPC_HAVE_IFADDRS + +#include "src/core/lib/iomgr/tcp_server_utils_posix.h" + +#include <errno.h> +#include <ifaddrs.h> +#include <stddef.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +#include "src/core/lib/iomgr/error.h" +#include "src/core/lib/iomgr/sockaddr.h" +#include "src/core/lib/iomgr/sockaddr_utils.h" + +/* Return the listener in s with address addr or NULL. */ +static grpc_tcp_listener *find_listener_with_addr(grpc_tcp_server *s, + grpc_resolved_address *addr) { + grpc_tcp_listener *l; + gpr_mu_lock(&s->mu); + for (l = s->head; l != NULL; l = l->next) { + if (l->addr.len != addr->len) { + continue; + } + if (memcmp(l->addr.addr, addr->addr, addr->len) == 0) { + break; + } + } + gpr_mu_unlock(&s->mu); + return l; +} + +/* Bind to "::" to get a port number not used by any address. */ +static grpc_error *get_unused_port(int *port) { + grpc_resolved_address wild; + grpc_sockaddr_make_wildcard6(0, &wild); + grpc_dualstack_mode dsmode; + int fd; + grpc_error *err = + grpc_create_dualstack_socket(&wild, SOCK_STREAM, 0, &dsmode, &fd); + if (err != GRPC_ERROR_NONE) { + return err; + } + if (dsmode == GRPC_DSMODE_IPV4) { + grpc_sockaddr_make_wildcard4(0, &wild); + } + if (bind(fd, (const struct sockaddr *)wild.addr, (socklen_t)wild.len) != 0) { + err = GRPC_OS_ERROR(errno, "bind"); + close(fd); + return err; + } + if (getsockname(fd, (struct sockaddr *)wild.addr, (socklen_t *)&wild.len) != + 0) { + err = GRPC_OS_ERROR(errno, "getsockname"); + close(fd); + return err; + } + close(fd); + *port = grpc_sockaddr_get_port(&wild); + return *port <= 0 ? GRPC_ERROR_CREATE("Bad port") : GRPC_ERROR_NONE; +} + +grpc_error *grpc_tcp_server_add_all_local_addrs(grpc_tcp_server *s, + unsigned port_index, + int requested_port, + int *out_port) { + struct ifaddrs *ifa = NULL; + struct ifaddrs *ifa_it; + unsigned fd_index = 0; + grpc_tcp_listener *sp = NULL; + grpc_error *err = GRPC_ERROR_NONE; + if (requested_port == 0) { + /* Note: There could be a race where some local addrs can listen on the + selected port and some can't. The sane way to handle this would be to + retry by recreating the whole grpc_tcp_server. Backing out individual + listeners and orphaning the FDs looks like too much trouble. */ + if ((err = get_unused_port(&requested_port)) != GRPC_ERROR_NONE) { + return err; + } else if (requested_port <= 0) { + return GRPC_ERROR_CREATE("Bad get_unused_port()"); + } + gpr_log(GPR_DEBUG, "Picked unused port %d", requested_port); + } + if (getifaddrs(&ifa) != 0 || ifa == NULL) { + return GRPC_OS_ERROR(errno, "getifaddrs"); + } + for (ifa_it = ifa; ifa_it != NULL; ifa_it = ifa_it->ifa_next) { + grpc_resolved_address addr; + char *addr_str = NULL; + grpc_dualstack_mode dsmode; + grpc_tcp_listener *new_sp = NULL; + const char *ifa_name = (ifa_it->ifa_name ? ifa_it->ifa_name : "<unknown>"); + if (ifa_it->ifa_addr == NULL) { + continue; + } else if (ifa_it->ifa_addr->sa_family == AF_INET) { + addr.len = sizeof(struct sockaddr_in); + } else if (ifa_it->ifa_addr->sa_family == AF_INET6) { + addr.len = sizeof(struct sockaddr_in6); + } else { + continue; + } + memcpy(addr.addr, ifa_it->ifa_addr, addr.len); + if (!grpc_sockaddr_set_port(&addr, requested_port)) { + /* Should never happen, because we check sa_family above. */ + err = GRPC_ERROR_CREATE("Failed to set port"); + break; + } + if (grpc_sockaddr_to_string(&addr_str, &addr, 0) < 0) { + addr_str = gpr_strdup("<error>"); + } + gpr_log(GPR_DEBUG, + "Adding local addr from interface %s flags 0x%x to server: %s", + ifa_name, ifa_it->ifa_flags, addr_str); + /* We could have multiple interfaces with the same address (e.g., bonding), + so look for duplicates. */ + if (find_listener_with_addr(s, &addr) != NULL) { + gpr_log(GPR_DEBUG, "Skipping duplicate addr %s on interface %s", addr_str, + ifa_name); + gpr_free(addr_str); + continue; + } + if ((err = grpc_tcp_server_add_addr(s, &addr, port_index, fd_index, &dsmode, + &new_sp)) != GRPC_ERROR_NONE) { + char *err_str = NULL; + grpc_error *root_err; + if (gpr_asprintf(&err_str, "Failed to add listener: %s", addr_str) < 0) { + err_str = gpr_strdup("Failed to add listener"); + } + root_err = GRPC_ERROR_CREATE(err_str); + gpr_free(err_str); + gpr_free(addr_str); + err = grpc_error_add_child(root_err, err); + break; + } else { + GPR_ASSERT(requested_port == new_sp->port); + ++fd_index; + if (sp != NULL) { + new_sp->is_sibling = 1; + sp->sibling = new_sp; + } + sp = new_sp; + } + gpr_free(addr_str); + } + freeifaddrs(ifa); + if (err != GRPC_ERROR_NONE) { + return err; + } else if (sp == NULL) { + return GRPC_ERROR_CREATE("No local addresses"); + } else { + *out_port = sp->port; + return GRPC_ERROR_NONE; + } +} + +bool grpc_tcp_server_have_ifaddrs(void) { return true; } + +#endif /* GRPC_HAVE_IFADDRS */ diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.c b/src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.c new file mode 100644 index 0000000000..95c3198be6 --- /dev/null +++ b/src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.c @@ -0,0 +1,49 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/iomgr/port.h" + +#if defined(GRPC_POSIX_SOCKET) && !defined(GRPC_HAVE_IFADDRS) + +#include "src/core/lib/iomgr/tcp_server_utils_posix.h" + +grpc_error *grpc_tcp_server_add_all_local_addrs(grpc_tcp_server *s, + unsigned port_index, + int requested_port, + int *out_port) { + return GRPC_ERROR_CREATE("no ifaddrs available"); +} + +bool grpc_tcp_server_have_ifaddrs(void) { return false; } + +#endif /* defined(GRPC_POSIX_SOCKET) && !defined(GRPC_HAVE_IFADDRS) */ diff --git a/src/core/lib/iomgr/timer_generic.c b/src/core/lib/iomgr/timer_generic.c index 6d638bcbaa..d4df96c214 100644 --- a/src/core/lib/iomgr/timer_generic.c +++ b/src/core/lib/iomgr/timer_generic.c @@ -42,6 +42,7 @@ #include <grpc/support/useful.h> #include "src/core/lib/iomgr/time_averaged_stats.h" #include "src/core/lib/iomgr/timer_heap.h" +#include "src/core/lib/support/spinlock.h" #define INVALID_HEAP_INDEX 0xffffffffu @@ -69,7 +70,7 @@ typedef struct { /* Protects g_shard_queue */ static gpr_mu g_mu; /* Allow only one run_some_expired_timers at once */ -static gpr_mu g_checker_mu; +static gpr_spinlock g_checker_mu = GPR_SPINLOCK_STATIC_INITIALIZER; static gpr_clock_type g_clock_type; static shard_type g_shards[NUM_SHARDS]; /* Protected by g_mu */ @@ -90,7 +91,6 @@ void grpc_timer_list_init(gpr_timespec now) { g_initialized = true; gpr_mu_init(&g_mu); - gpr_mu_init(&g_checker_mu); g_clock_type = now.clock_type; for (i = 0; i < NUM_SHARDS; i++) { @@ -117,7 +117,6 @@ void grpc_timer_list_shutdown(grpc_exec_ctx *exec_ctx) { grpc_timer_heap_destroy(&shard->heap); } gpr_mu_destroy(&g_mu); - gpr_mu_destroy(&g_checker_mu); g_initialized = false; } @@ -324,7 +323,7 @@ static int run_some_expired_timers(grpc_exec_ctx *exec_ctx, gpr_timespec now, /* TODO(ctiller): verify that there are any timers (atomically) here */ - if (gpr_mu_trylock(&g_checker_mu)) { + if (gpr_spinlock_trylock(&g_checker_mu)) { gpr_mu_lock(&g_mu); while (gpr_time_cmp(g_shard_queue[0]->min_deadline, now) < 0) { @@ -350,7 +349,7 @@ static int run_some_expired_timers(grpc_exec_ctx *exec_ctx, gpr_timespec now, } gpr_mu_unlock(&g_mu); - gpr_mu_unlock(&g_checker_mu); + gpr_spinlock_unlock(&g_checker_mu); } else if (next != NULL) { /* TODO(ctiller): this forces calling code to do an short poll, and then retry the timer check (because this time through the timer list was diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index 2a1c8d39fa..71e295770a 100644 --- a/src/core/lib/iomgr/udp_server.c +++ b/src/core/lib/iomgr/udp_server.c @@ -109,8 +109,8 @@ struct grpc_udp_server { grpc_pollset **pollsets; /* number of pollsets in the pollsets array */ size_t pollset_count; - /* The parent grpc server */ - grpc_server *grpc_server; + /* opaque object to pass to callbacks */ + void *user_data; }; grpc_udp_server *grpc_udp_server_create(void) { @@ -178,7 +178,7 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { /* Call the orphan_cb to signal that the FD is about to be closed and * should no longer be used. */ GPR_ASSERT(sp->orphan_cb); - sp->orphan_cb(exec_ctx, sp->emfd); + sp->orphan_cb(exec_ctx, sp->emfd, sp->server->user_data); grpc_fd_orphan(exec_ctx, sp->emfd, &sp->destroyed_closure, NULL, "udp_listener_shutdown"); @@ -204,7 +204,7 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, if (s->active_ports) { for (sp = s->head; sp; sp = sp->next) { GPR_ASSERT(sp->orphan_cb); - sp->orphan_cb(exec_ctx, sp->emfd); + sp->orphan_cb(exec_ctx, sp->emfd, sp->server->user_data); grpc_fd_shutdown(exec_ctx, sp->emfd, GRPC_ERROR_CREATE("Server destroyed")); } @@ -299,7 +299,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { /* Tell the registered callback that data is available to read. */ GPR_ASSERT(sp->read_cb); - sp->read_cb(exec_ctx, sp->emfd, sp->server->grpc_server); + sp->read_cb(exec_ctx, sp->emfd, sp->server->user_data); /* Re-arm the notification event so we get another chance to read. */ grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure); @@ -322,7 +322,7 @@ static void on_write(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { /* Tell the registered callback that the socket is writeable. */ GPR_ASSERT(sp->write_cb); - sp->write_cb(exec_ctx, sp->emfd); + sp->write_cb(exec_ctx, sp->emfd, sp->server->user_data); /* Re-arm the notification event so we get another chance to write. */ grpc_fd_notify_on_write(exec_ctx, sp->emfd, &sp->write_closure); @@ -464,13 +464,13 @@ int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index) { void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, grpc_pollset **pollsets, size_t pollset_count, - grpc_server *server) { + void *user_data) { size_t i; gpr_mu_lock(&s->mu); grpc_udp_listener *sp; GPR_ASSERT(s->active_ports == 0); s->pollsets = pollsets; - s->grpc_server = server; + s->user_data = user_data; sp = s->head; while (sp != NULL) { @@ -485,7 +485,11 @@ void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, grpc_schedule_on_exec_ctx); grpc_fd_notify_on_write(exec_ctx, sp->emfd, &sp->write_closure); - s->active_ports++; + /* Registered for both read and write callbacks: increment active_ports + * twice to account for this, and delay free-ing of memory until both + * on_read and on_write have fired. */ + s->active_ports += 2; + sp = sp->next; } diff --git a/src/core/lib/iomgr/udp_server.h b/src/core/lib/iomgr/udp_server.h index ed63fa7d81..90842a47f0 100644 --- a/src/core/lib/iomgr/udp_server.h +++ b/src/core/lib/iomgr/udp_server.h @@ -47,23 +47,23 @@ typedef struct grpc_udp_server grpc_udp_server; /* Called when data is available to read from the socket. */ typedef void (*grpc_udp_server_read_cb)(grpc_exec_ctx *exec_ctx, grpc_fd *emfd, - struct grpc_server *server); + void *user_data); /* Called when the socket is writeable. */ -typedef void (*grpc_udp_server_write_cb)(grpc_exec_ctx *exec_ctx, - grpc_fd *emfd); +typedef void (*grpc_udp_server_write_cb)(grpc_exec_ctx *exec_ctx, grpc_fd *emfd, + void *user_data); /* Called when the grpc_fd is about to be orphaned (and the FD closed). */ typedef void (*grpc_udp_server_orphan_cb)(grpc_exec_ctx *exec_ctx, - grpc_fd *emfd); + grpc_fd *emfd, void *user_data); /* Create a server, initially not bound to any ports */ grpc_udp_server *grpc_udp_server_create(void); -/* Start listening to bound ports */ +/* Start listening to bound ports. user_data is passed to callbacks. */ void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *udp_server, grpc_pollset **pollsets, size_t pollset_count, - struct grpc_server *server); + void *user_data); int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index); diff --git a/src/core/lib/iomgr/wakeup_fd_posix.h b/src/core/lib/iomgr/wakeup_fd_posix.h index 71d32d97ba..c8dd242c75 100644 --- a/src/core/lib/iomgr/wakeup_fd_posix.h +++ b/src/core/lib/iomgr/wakeup_fd_posix.h @@ -46,7 +46,7 @@ * * Setup: * 1. Before calling anything, call global_init() at least once. - * 1. Call grpc_wakeup_fd_create() to get a wakeup_fd. + * 1. Call grpc_wakeup_fd_init() to set up a wakeup_fd. * 2. Add the result of GRPC_WAKEUP_FD_FD to the set of monitored file * descriptors for the poll() style API you are using. Monitor the file * descriptor for readability. diff --git a/src/core/lib/json/json.c b/src/core/lib/json/json.c index 48b13686d7..5f7e4ec042 100644 --- a/src/core/lib/json/json.c +++ b/src/core/lib/json/json.c @@ -38,8 +38,7 @@ #include "src/core/lib/json/json.h" grpc_json* grpc_json_create(grpc_json_type type) { - grpc_json* json = gpr_malloc(sizeof(*json)); - memset(json, 0, sizeof(*json)); + grpc_json* json = gpr_zalloc(sizeof(*json)); json->type = type; return json; diff --git a/src/core/lib/security/context/security_context.c b/src/core/lib/security/context/security_context.c index fe82fabc31..d29fc55ea0 100644 --- a/src/core/lib/security/context/security_context.c +++ b/src/core/lib/security/context/security_context.c @@ -91,10 +91,7 @@ void grpc_auth_context_release(grpc_auth_context *context) { /* --- grpc_client_security_context --- */ grpc_client_security_context *grpc_client_security_context_create(void) { - grpc_client_security_context *ctx = - gpr_malloc(sizeof(grpc_client_security_context)); - memset(ctx, 0, sizeof(grpc_client_security_context)); - return ctx; + return gpr_zalloc(sizeof(grpc_client_security_context)); } void grpc_client_security_context_destroy(void *ctx) { @@ -112,10 +109,7 @@ void grpc_client_security_context_destroy(void *ctx) { /* --- grpc_server_security_context --- */ grpc_server_security_context *grpc_server_security_context_create(void) { - grpc_server_security_context *ctx = - gpr_malloc(sizeof(grpc_server_security_context)); - memset(ctx, 0, sizeof(grpc_server_security_context)); - return ctx; + return gpr_zalloc(sizeof(grpc_server_security_context)); } void grpc_server_security_context_destroy(void *ctx) { @@ -132,8 +126,7 @@ void grpc_server_security_context_destroy(void *ctx) { static grpc_auth_property_iterator empty_iterator = {NULL, 0, NULL}; grpc_auth_context *grpc_auth_context_create(grpc_auth_context *chained) { - grpc_auth_context *ctx = gpr_malloc(sizeof(grpc_auth_context)); - memset(ctx, 0, sizeof(grpc_auth_context)); + grpc_auth_context *ctx = gpr_zalloc(sizeof(grpc_auth_context)); gpr_ref_init(&ctx->refcount, 1); if (chained != NULL) { ctx->chained = GRPC_AUTH_CONTEXT_REF(chained, "chained"); diff --git a/src/core/lib/security/credentials/composite/composite_credentials.c b/src/core/lib/security/credentials/composite/composite_credentials.c index be1588dd8b..e497b9f657 100644 --- a/src/core/lib/security/credentials/composite/composite_credentials.c +++ b/src/core/lib/security/credentials/composite/composite_credentials.c @@ -115,8 +115,7 @@ static void composite_call_get_request_metadata( grpc_composite_call_credentials *c = (grpc_composite_call_credentials *)creds; grpc_composite_call_credentials_metadata_context *ctx; - ctx = gpr_malloc(sizeof(grpc_composite_call_credentials_metadata_context)); - memset(ctx, 0, sizeof(grpc_composite_call_credentials_metadata_context)); + ctx = gpr_zalloc(sizeof(grpc_composite_call_credentials_metadata_context)); ctx->auth_md_context = auth_md_context; ctx->user_data = user_data; ctx->cb = cb; @@ -158,8 +157,7 @@ grpc_call_credentials *grpc_composite_call_credentials_create( GPR_ASSERT(reserved == NULL); GPR_ASSERT(creds1 != NULL); GPR_ASSERT(creds2 != NULL); - c = gpr_malloc(sizeof(grpc_composite_call_credentials)); - memset(c, 0, sizeof(grpc_composite_call_credentials)); + c = gpr_zalloc(sizeof(grpc_composite_call_credentials)); c->base.type = GRPC_CALL_CREDENTIALS_TYPE_COMPOSITE; c->base.vtable = &composite_call_credentials_vtable; gpr_ref_init(&c->base.refcount, 1); @@ -167,8 +165,7 @@ grpc_call_credentials *grpc_composite_call_credentials_create( creds2_array = get_creds_array(&creds2); c->inner.num_creds = creds1_array.num_creds + creds2_array.num_creds; creds_array_byte_size = c->inner.num_creds * sizeof(grpc_call_credentials *); - c->inner.creds_array = gpr_malloc(creds_array_byte_size); - memset(c->inner.creds_array, 0, creds_array_byte_size); + c->inner.creds_array = gpr_zalloc(creds_array_byte_size); for (i = 0; i < creds1_array.num_creds; i++) { grpc_call_credentials *cur_creds = creds1_array.creds_array[i]; c->inner.creds_array[i] = grpc_call_credentials_ref(cur_creds); @@ -262,8 +259,7 @@ static grpc_channel_credentials_vtable composite_channel_credentials_vtable = { grpc_channel_credentials *grpc_composite_channel_credentials_create( grpc_channel_credentials *channel_creds, grpc_call_credentials *call_creds, void *reserved) { - grpc_composite_channel_credentials *c = gpr_malloc(sizeof(*c)); - memset(c, 0, sizeof(*c)); + grpc_composite_channel_credentials *c = gpr_zalloc(sizeof(*c)); GPR_ASSERT(channel_creds != NULL && call_creds != NULL && reserved == NULL); GRPC_API_TRACE( "grpc_composite_channel_credentials_create(channel_creds=%p, " diff --git a/src/core/lib/security/credentials/credentials.c b/src/core/lib/security/credentials/credentials.c index b24697ce54..52b80141d1 100644 --- a/src/core/lib/security/credentials/credentials.c +++ b/src/core/lib/security/credentials/credentials.c @@ -57,8 +57,7 @@ grpc_credentials_metadata_request *grpc_credentials_metadata_request_create( grpc_call_credentials *creds, grpc_credentials_metadata_cb cb, void *user_data) { grpc_credentials_metadata_request *r = - gpr_malloc(sizeof(grpc_credentials_metadata_request)); - memset(&r->response, 0, sizeof(r->response)); + gpr_zalloc(sizeof(grpc_credentials_metadata_request)); r->creds = grpc_call_credentials_ref(creds); r->cb = cb; r->user_data = user_data; diff --git a/src/core/lib/security/credentials/credentials_metadata.c b/src/core/lib/security/credentials/credentials_metadata.c index 68da5fb4a8..f11cc58786 100644 --- a/src/core/lib/security/credentials/credentials_metadata.c +++ b/src/core/lib/security/credentials/credentials_metadata.c @@ -50,8 +50,7 @@ static void store_ensure_capacity(grpc_credentials_md_store *store) { grpc_credentials_md_store *grpc_credentials_md_store_create( size_t initial_capacity) { grpc_credentials_md_store *store = - gpr_malloc(sizeof(grpc_credentials_md_store)); - memset(store, 0, sizeof(grpc_credentials_md_store)); + gpr_zalloc(sizeof(grpc_credentials_md_store)); if (initial_capacity > 0) { store->entries = gpr_malloc(initial_capacity * sizeof(grpc_credentials_md)); store->allocated = initial_capacity; diff --git a/src/core/lib/security/credentials/fake/fake_credentials.c b/src/core/lib/security/credentials/fake/fake_credentials.c index a0629f76ce..68636ba208 100644 --- a/src/core/lib/security/credentials/fake/fake_credentials.c +++ b/src/core/lib/security/credentials/fake/fake_credentials.c @@ -71,8 +71,7 @@ static grpc_server_credentials_vtable grpc_channel_credentials *grpc_fake_transport_security_credentials_create( void) { - grpc_channel_credentials *c = gpr_malloc(sizeof(grpc_channel_credentials)); - memset(c, 0, sizeof(grpc_channel_credentials)); + grpc_channel_credentials *c = gpr_zalloc(sizeof(grpc_channel_credentials)); c->type = GRPC_CHANNEL_CREDENTIALS_TYPE_FAKE_TRANSPORT_SECURITY; c->vtable = &fake_transport_security_credentials_vtable; gpr_ref_init(&c->refcount, 1); @@ -131,8 +130,7 @@ static grpc_call_credentials_vtable md_only_test_vtable = { grpc_call_credentials *grpc_md_only_test_credentials_create( const char *md_key, const char *md_value, int is_async) { grpc_md_only_test_credentials *c = - gpr_malloc(sizeof(grpc_md_only_test_credentials)); - memset(c, 0, sizeof(grpc_md_only_test_credentials)); + gpr_zalloc(sizeof(grpc_md_only_test_credentials)); c->base.type = GRPC_CALL_CREDENTIALS_TYPE_OAUTH2; c->base.vtable = &md_only_test_vtable; gpr_ref_init(&c->base.refcount, 1); diff --git a/src/core/lib/security/credentials/google_default/google_default_credentials.c b/src/core/lib/security/credentials/google_default/google_default_credentials.c index ecd26de9fa..dd44621347 100644 --- a/src/core/lib/security/credentials/google_default/google_default_credentials.c +++ b/src/core/lib/security/credentials/google_default/google_default_credentials.c @@ -112,7 +112,7 @@ static int is_stack_running_on_compute_engine(grpc_exec_ctx *exec_ctx) { on compute engine. */ gpr_timespec max_detection_delay = gpr_time_from_seconds(1, GPR_TIMESPAN); - grpc_pollset *pollset = gpr_malloc(grpc_pollset_size()); + grpc_pollset *pollset = gpr_zalloc(grpc_pollset_size()); grpc_pollset_init(pollset, &g_polling_mu); detector.pollent = grpc_polling_entity_create_from_pollset(pollset); detector.is_done = 0; diff --git a/src/core/lib/security/credentials/iam/iam_credentials.c b/src/core/lib/security/credentials/iam/iam_credentials.c index abd69a9670..393a433f2e 100644 --- a/src/core/lib/security/credentials/iam/iam_credentials.c +++ b/src/core/lib/security/credentials/iam/iam_credentials.c @@ -72,8 +72,7 @@ grpc_call_credentials *grpc_google_iam_credentials_create( GPR_ASSERT(reserved == NULL); GPR_ASSERT(token != NULL); GPR_ASSERT(authority_selector != NULL); - c = gpr_malloc(sizeof(grpc_google_iam_credentials)); - memset(c, 0, sizeof(grpc_google_iam_credentials)); + c = gpr_zalloc(sizeof(grpc_google_iam_credentials)); c->base.type = GRPC_CALL_CREDENTIALS_TYPE_IAM; c->base.vtable = &iam_vtable; gpr_ref_init(&c->base.refcount, 1); diff --git a/src/core/lib/security/credentials/jwt/jwt_credentials.c b/src/core/lib/security/credentials/jwt/jwt_credentials.c index 616be64a54..178ce89aa6 100644 --- a/src/core/lib/security/credentials/jwt/jwt_credentials.c +++ b/src/core/lib/security/credentials/jwt/jwt_credentials.c @@ -135,8 +135,7 @@ grpc_service_account_jwt_access_credentials_create_from_auth_json_key( gpr_log(GPR_ERROR, "Invalid input for jwt credentials creation"); return NULL; } - c = gpr_malloc(sizeof(grpc_service_account_jwt_access_credentials)); - memset(c, 0, sizeof(grpc_service_account_jwt_access_credentials)); + c = gpr_zalloc(sizeof(grpc_service_account_jwt_access_credentials)); c->base.type = GRPC_CALL_CREDENTIALS_TYPE_JWT; gpr_ref_init(&c->base.refcount, 1); c->base.vtable = &jwt_vtable; diff --git a/src/core/lib/security/credentials/jwt/jwt_verifier.c b/src/core/lib/security/credentials/jwt/jwt_verifier.c index f128177e8c..5c59cf0f4a 100644 --- a/src/core/lib/security/credentials/jwt/jwt_verifier.c +++ b/src/core/lib/security/credentials/jwt/jwt_verifier.c @@ -144,8 +144,7 @@ static void jose_header_destroy(grpc_exec_ctx *exec_ctx, jose_header *h) { static jose_header *jose_header_from_json(grpc_exec_ctx *exec_ctx, grpc_json *json, grpc_slice buffer) { grpc_json *cur; - jose_header *h = gpr_malloc(sizeof(jose_header)); - memset(h, 0, sizeof(jose_header)); + jose_header *h = gpr_zalloc(sizeof(jose_header)); h->buffer = buffer; for (cur = json->child; cur != NULL; cur = cur->next) { if (strcmp(cur->key, "alg") == 0) { @@ -363,8 +362,7 @@ static verifier_cb_ctx *verifier_cb_ctx_create( const char *signed_jwt, size_t signed_jwt_len, void *user_data, grpc_jwt_verification_done_cb cb) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - verifier_cb_ctx *ctx = gpr_malloc(sizeof(verifier_cb_ctx)); - memset(ctx, 0, sizeof(verifier_cb_ctx)); + verifier_cb_ctx *ctx = gpr_zalloc(sizeof(verifier_cb_ctx)); ctx->verifier = verifier; ctx->pollent = grpc_polling_entity_create_from_pollset(pollset); ctx->header = header; @@ -878,8 +876,7 @@ error: grpc_jwt_verifier *grpc_jwt_verifier_create( const grpc_jwt_verifier_email_domain_key_url_mapping *mappings, size_t num_mappings) { - grpc_jwt_verifier *v = gpr_malloc(sizeof(grpc_jwt_verifier)); - memset(v, 0, sizeof(grpc_jwt_verifier)); + grpc_jwt_verifier *v = gpr_zalloc(sizeof(grpc_jwt_verifier)); grpc_httpcli_context_init(&v->http_ctx); /* We know at least of one mapping. */ diff --git a/src/core/lib/security/credentials/oauth2/oauth2_credentials.c b/src/core/lib/security/credentials/oauth2/oauth2_credentials.c index c0f260f938..ccfb3566c1 100644 --- a/src/core/lib/security/credentials/oauth2/oauth2_credentials.c +++ b/src/core/lib/security/credentials/oauth2/oauth2_credentials.c @@ -389,8 +389,7 @@ grpc_refresh_token_credentials_create_from_auth_refresh_token( gpr_log(GPR_ERROR, "Invalid input for refresh token credentials creation"); return NULL; } - c = gpr_malloc(sizeof(grpc_google_refresh_token_credentials)); - memset(c, 0, sizeof(grpc_google_refresh_token_credentials)); + c = gpr_zalloc(sizeof(grpc_google_refresh_token_credentials)); init_oauth2_token_fetcher(&c->base, refresh_token_fetch_oauth2); c->base.base.vtable = &refresh_token_vtable; c->refresh_token = refresh_token; @@ -450,14 +449,13 @@ static grpc_call_credentials_vtable access_token_vtable = { grpc_call_credentials *grpc_access_token_credentials_create( const char *access_token, void *reserved) { grpc_access_token_credentials *c = - gpr_malloc(sizeof(grpc_access_token_credentials)); + gpr_zalloc(sizeof(grpc_access_token_credentials)); char *token_md_value; GRPC_API_TRACE( "grpc_access_token_credentials_create(access_token=<redacted>, " "reserved=%p)", 1, (reserved)); GPR_ASSERT(reserved == NULL); - memset(c, 0, sizeof(grpc_access_token_credentials)); c->base.type = GRPC_CALL_CREDENTIALS_TYPE_OAUTH2; c->base.vtable = &access_token_vtable; gpr_ref_init(&c->base.refcount, 1); diff --git a/src/core/lib/security/credentials/plugin/plugin_credentials.c b/src/core/lib/security/credentials/plugin/plugin_credentials.c index 7bc5dfb403..8416bfa5a8 100644 --- a/src/core/lib/security/credentials/plugin/plugin_credentials.c +++ b/src/core/lib/security/credentials/plugin/plugin_credentials.c @@ -126,8 +126,7 @@ static void plugin_get_request_metadata(grpc_exec_ctx *exec_ctx, void *user_data) { grpc_plugin_credentials *c = (grpc_plugin_credentials *)creds; if (c->plugin.get_metadata != NULL) { - grpc_metadata_plugin_request *request = gpr_malloc(sizeof(*request)); - memset(request, 0, sizeof(*request)); + grpc_metadata_plugin_request *request = gpr_zalloc(sizeof(*request)); request->user_data = user_data; request->cb = cb; c->plugin.get_metadata(c->plugin.state, context, @@ -142,11 +141,10 @@ static grpc_call_credentials_vtable plugin_vtable = { grpc_call_credentials *grpc_metadata_credentials_create_from_plugin( grpc_metadata_credentials_plugin plugin, void *reserved) { - grpc_plugin_credentials *c = gpr_malloc(sizeof(*c)); + grpc_plugin_credentials *c = gpr_zalloc(sizeof(*c)); GRPC_API_TRACE("grpc_metadata_credentials_create_from_plugin(reserved=%p)", 1, (reserved)); GPR_ASSERT(reserved == NULL); - memset(c, 0, sizeof(*c)); c->base.type = plugin.type; c->base.vtable = &plugin_vtable; gpr_ref_init(&c->base.refcount, 1); diff --git a/src/core/lib/security/credentials/ssl/ssl_credentials.c b/src/core/lib/security/credentials/ssl/ssl_credentials.c index 4eebb7d613..4b17ac8098 100644 --- a/src/core/lib/security/credentials/ssl/ssl_credentials.c +++ b/src/core/lib/security/credentials/ssl/ssl_credentials.c @@ -121,14 +121,13 @@ static void ssl_build_config(const char *pem_root_certs, grpc_channel_credentials *grpc_ssl_credentials_create( const char *pem_root_certs, grpc_ssl_pem_key_cert_pair *pem_key_cert_pair, void *reserved) { - grpc_ssl_credentials *c = gpr_malloc(sizeof(grpc_ssl_credentials)); + grpc_ssl_credentials *c = gpr_zalloc(sizeof(grpc_ssl_credentials)); GRPC_API_TRACE( "grpc_ssl_credentials_create(pem_root_certs=%s, " "pem_key_cert_pair=%p, " "reserved=%p)", 3, (pem_root_certs, pem_key_cert_pair, reserved)); GPR_ASSERT(reserved == NULL); - memset(c, 0, sizeof(grpc_ssl_credentials)); c->base.type = GRPC_CHANNEL_CREDENTIALS_TYPE_SSL; c->base.vtable = &ssl_vtable; gpr_ref_init(&c->base.refcount, 1); @@ -225,7 +224,7 @@ grpc_server_credentials *grpc_ssl_server_credentials_create_ex( grpc_ssl_client_certificate_request_type client_certificate_request, void *reserved) { grpc_ssl_server_credentials *c = - gpr_malloc(sizeof(grpc_ssl_server_credentials)); + gpr_zalloc(sizeof(grpc_ssl_server_credentials)); GRPC_API_TRACE( "grpc_ssl_server_credentials_create_ex(" "pem_root_certs=%s, pem_key_cert_pairs=%p, num_key_cert_pairs=%lu, " @@ -233,7 +232,6 @@ grpc_server_credentials *grpc_ssl_server_credentials_create_ex( 5, (pem_root_certs, pem_key_cert_pairs, (unsigned long)num_key_cert_pairs, client_certificate_request, reserved)); GPR_ASSERT(reserved == NULL); - memset(c, 0, sizeof(grpc_ssl_server_credentials)); c->base.type = GRPC_CHANNEL_CREDENTIALS_TYPE_SSL; gpr_ref_init(&c->base.refcount, 1); c->base.vtable = &ssl_server_vtable; diff --git a/src/core/lib/security/transport/client_auth_filter.c b/src/core/lib/security/transport/client_auth_filter.c index a23082a866..8dea1d98ff 100644 --- a/src/core/lib/security/transport/client_auth_filter.c +++ b/src/core/lib/security/transport/client_auth_filter.c @@ -318,7 +318,7 @@ static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *ignored) { + grpc_closure *ignored) { call_data *calld = elem->call_data; grpc_call_credentials_unref(exec_ctx, calld->creds); if (calld->have_host) { diff --git a/src/core/lib/security/transport/security_connector.c b/src/core/lib/security/transport/security_connector.c index aeb04e33a3..ad083a730f 100644 --- a/src/core/lib/security/transport/security_connector.c +++ b/src/core/lib/security/transport/security_connector.c @@ -412,8 +412,7 @@ static grpc_security_connector_vtable fake_server_vtable = { grpc_channel_security_connector *grpc_fake_channel_security_connector_create( grpc_call_credentials *request_metadata_creds, const char *target, const grpc_channel_args *args) { - grpc_fake_channel_security_connector *c = gpr_malloc(sizeof(*c)); - memset(c, 0, sizeof(*c)); + grpc_fake_channel_security_connector *c = gpr_zalloc(sizeof(*c)); gpr_ref_init(&c->base.base.refcount, 1); c->base.base.url_scheme = GRPC_FAKE_SECURITY_URL_SCHEME; c->base.base.vtable = &fake_channel_vtable; @@ -435,8 +434,7 @@ grpc_channel_security_connector *grpc_fake_channel_security_connector_create( grpc_server_security_connector *grpc_fake_server_security_connector_create( void) { grpc_server_security_connector *c = - gpr_malloc(sizeof(grpc_server_security_connector)); - memset(c, 0, sizeof(*c)); + gpr_zalloc(sizeof(grpc_server_security_connector)); gpr_ref_init(&c->base.refcount, 1); c->base.vtable = &fake_server_vtable; c->base.url_scheme = GRPC_FAKE_SECURITY_URL_SCHEME; @@ -815,8 +813,7 @@ grpc_security_status grpc_ssl_channel_security_connector_create( pem_root_certs_size = config->pem_root_certs_size; } - c = gpr_malloc(sizeof(grpc_ssl_channel_security_connector)); - memset(c, 0, sizeof(grpc_ssl_channel_security_connector)); + c = gpr_zalloc(sizeof(grpc_ssl_channel_security_connector)); gpr_ref_init(&c->base.base.refcount, 1); c->base.base.vtable = &ssl_channel_vtable; @@ -877,8 +874,7 @@ grpc_security_status grpc_ssl_server_security_connector_create( gpr_log(GPR_ERROR, "An SSL server needs a key and a cert."); goto error; } - c = gpr_malloc(sizeof(grpc_ssl_server_security_connector)); - memset(c, 0, sizeof(grpc_ssl_server_security_connector)); + c = gpr_zalloc(sizeof(grpc_ssl_server_security_connector)); gpr_ref_init(&c->base.base.refcount, 1); c->base.base.url_scheme = GRPC_SSL_URL_SCHEME; diff --git a/src/core/lib/security/transport/security_handshaker.c b/src/core/lib/security/transport/security_handshaker.c index 5d57543ac5..7065d261ba 100644 --- a/src/core/lib/security/transport/security_handshaker.c +++ b/src/core/lib/security/transport/security_handshaker.c @@ -387,8 +387,7 @@ static const grpc_handshaker_vtable security_handshaker_vtable = { static grpc_handshaker *security_handshaker_create( grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker, grpc_security_connector *connector) { - security_handshaker *h = gpr_malloc(sizeof(security_handshaker)); - memset(h, 0, sizeof(security_handshaker)); + security_handshaker *h = gpr_zalloc(sizeof(security_handshaker)); grpc_handshaker_init(&security_handshaker_vtable, &h->base); h->handshaker = handshaker; h->connector = GRPC_SECURITY_CONNECTOR_REF(connector, "handshake"); diff --git a/src/core/lib/security/transport/server_auth_filter.c b/src/core/lib/security/transport/server_auth_filter.c index 14619d97ca..01cb473177 100644 --- a/src/core/lib/security/transport/server_auth_filter.c +++ b/src/core/lib/security/transport/server_auth_filter.c @@ -227,7 +227,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *ignored) {} + grpc_closure *ignored) {} /* Constructor for channel_data */ static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/slice/slice_hash_table.c b/src/core/lib/slice/slice_hash_table.c index 46f807f4a5..219567f36f 100644 --- a/src/core/lib/slice/slice_hash_table.c +++ b/src/core/lib/slice/slice_hash_table.c @@ -82,15 +82,13 @@ static void grpc_slice_hash_table_add( grpc_slice_hash_table* grpc_slice_hash_table_create( size_t num_entries, grpc_slice_hash_table_entry* entries) { - grpc_slice_hash_table* table = gpr_malloc(sizeof(*table)); - memset(table, 0, sizeof(*table)); + grpc_slice_hash_table* table = gpr_zalloc(sizeof(*table)); gpr_ref_init(&table->refs, 1); // Quadratic probing gets best performance when the table is no more // than half full. table->size = num_entries * 2; const size_t entry_size = sizeof(grpc_slice_hash_table_entry) * table->size; - table->entries = gpr_malloc(entry_size); - memset(table->entries, 0, entry_size); + table->entries = gpr_zalloc(entry_size); for (size_t i = 0; i < num_entries; ++i) { grpc_slice_hash_table_entry* entry = &entries[i]; grpc_slice_hash_table_add(table, entry->key, entry->value, entry->vtable); diff --git a/src/core/lib/slice/slice_intern.c b/src/core/lib/slice/slice_intern.c index 32adc4df97..1c00a2d88e 100644 --- a/src/core/lib/slice/slice_intern.c +++ b/src/core/lib/slice/slice_intern.c @@ -144,8 +144,7 @@ static void grow_shard(slice_shard *shard) { GPR_TIMER_BEGIN("grow_strtab", 0); - strtab = gpr_malloc(sizeof(interned_slice_refcount *) * capacity); - memset(strtab, 0, sizeof(interned_slice_refcount *) * capacity); + strtab = gpr_zalloc(sizeof(interned_slice_refcount *) * capacity); for (i = 0; i < shard->capacity; i++) { for (s = shard->strs[i]; s; s = next) { @@ -296,8 +295,7 @@ void grpc_slice_intern_init(void) { gpr_mu_init(&shard->mu); shard->count = 0; shard->capacity = INITIAL_SHARD_CAPACITY; - shard->strs = gpr_malloc(sizeof(*shard->strs) * shard->capacity); - memset(shard->strs, 0, sizeof(*shard->strs) * shard->capacity); + shard->strs = gpr_zalloc(sizeof(*shard->strs) * shard->capacity); } for (size_t i = 0; i < GPR_ARRAY_SIZE(static_metadata_hash); i++) { static_metadata_hash[i].hash = 0; diff --git a/src/core/lib/support/alloc.c b/src/core/lib/support/alloc.c index 618c3f6acd..9973166217 100644 --- a/src/core/lib/support/alloc.c +++ b/src/core/lib/support/alloc.c @@ -36,9 +36,19 @@ #include <grpc/support/log.h> #include <grpc/support/port_platform.h> #include <stdlib.h> +#include <string.h> #include "src/core/lib/profiling/timers.h" -static gpr_allocation_functions g_alloc_functions = {malloc, realloc, free}; +static void *zalloc_with_calloc(size_t sz) { return calloc(sz, 1); } + +static void *zalloc_with_gpr_malloc(size_t sz) { + void *p = gpr_malloc(sz); + memset(p, 0, sz); + return p; +} + +static gpr_allocation_functions g_alloc_functions = {malloc, zalloc_with_calloc, + realloc, free}; gpr_allocation_functions gpr_get_allocation_functions() { return g_alloc_functions; @@ -48,6 +58,9 @@ void gpr_set_allocation_functions(gpr_allocation_functions functions) { GPR_ASSERT(functions.malloc_fn != NULL); GPR_ASSERT(functions.realloc_fn != NULL); GPR_ASSERT(functions.free_fn != NULL); + if (functions.zalloc_fn == NULL) { + functions.zalloc_fn = zalloc_with_gpr_malloc; + } g_alloc_functions = functions; } @@ -63,6 +76,18 @@ void *gpr_malloc(size_t size) { return p; } +void *gpr_zalloc(size_t size) { + void *p; + if (size == 0) return NULL; + GPR_TIMER_BEGIN("gpr_zalloc", 0); + p = g_alloc_functions.zalloc_fn(size); + if (!p) { + abort(); + } + GPR_TIMER_END("gpr_zalloc", 0); + return p; +} + void gpr_free(void *p) { GPR_TIMER_BEGIN("gpr_free", 0); g_alloc_functions.free_fn(p); diff --git a/src/core/lib/support/arena.c b/src/core/lib/support/arena.c new file mode 100644 index 0000000000..7bcb983f24 --- /dev/null +++ b/src/core/lib/support/arena.c @@ -0,0 +1,98 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/support/arena.h" +#include <grpc/support/alloc.h> +#include <grpc/support/atm.h> +#include <grpc/support/log.h> +#include <grpc/support/useful.h> + +#define ROUND_UP_TO_ALIGNMENT_SIZE(x) \ + (((x) + GPR_MAX_ALIGNMENT - 1u) & ~(GPR_MAX_ALIGNMENT - 1u)) + +typedef struct zone { + size_t size_begin; + size_t size_end; + gpr_atm next_atm; +} zone; + +struct gpr_arena { + gpr_atm size_so_far; + zone initial_zone; +}; + +gpr_arena *gpr_arena_create(size_t initial_size) { + initial_size = ROUND_UP_TO_ALIGNMENT_SIZE(initial_size); + gpr_arena *a = gpr_zalloc(sizeof(gpr_arena) + initial_size); + a->initial_zone.size_end = initial_size; + return a; +} + +size_t gpr_arena_destroy(gpr_arena *arena) { + gpr_atm size = gpr_atm_no_barrier_load(&arena->size_so_far); + zone *z = (zone *)gpr_atm_no_barrier_load(&arena->initial_zone.next_atm); + gpr_free(arena); + while (z) { + zone *next_z = (zone *)gpr_atm_no_barrier_load(&z->next_atm); + gpr_free(z); + z = next_z; + } + return (size_t)size; +} + +void *gpr_arena_alloc(gpr_arena *arena, size_t size) { + size = ROUND_UP_TO_ALIGNMENT_SIZE(size); + size_t start = + (size_t)gpr_atm_no_barrier_fetch_add(&arena->size_so_far, size); + zone *z = &arena->initial_zone; + while (start > z->size_end) { + zone *next_z = (zone *)gpr_atm_acq_load(&z->next_atm); + if (next_z == NULL) { + size_t next_z_size = (size_t)gpr_atm_no_barrier_load(&arena->size_so_far); + next_z = gpr_zalloc(sizeof(zone) + next_z_size); + next_z->size_begin = z->size_end; + next_z->size_end = z->size_end + next_z_size; + if (!gpr_atm_rel_cas(&z->next_atm, (gpr_atm)NULL, (gpr_atm)next_z)) { + gpr_free(next_z); + next_z = (zone *)gpr_atm_acq_load(&z->next_atm); + } + } + z = next_z; + } + if (start + size > z->size_end) { + return gpr_arena_alloc(arena, size); + } + GPR_ASSERT(start >= z->size_begin); + GPR_ASSERT(start + size <= z->size_end); + return ((char *)(z + 1)) + start - z->size_begin; +} diff --git a/src/core/lib/support/arena.h b/src/core/lib/support/arena.h new file mode 100644 index 0000000000..c28033ffc3 --- /dev/null +++ b/src/core/lib/support/arena.h @@ -0,0 +1,54 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +// \file Arena based allocator +// Allows very fast allocation of memory, but that memory cannot be freed until +// the arena as a whole is freed +// Tracks the total memory allocated against it, so that future arenas can +// pre-allocate the right amount of memory + +#ifndef GRPC_CORE_LIB_SUPPORT_ARENA_H +#define GRPC_CORE_LIB_SUPPORT_ARENA_H + +#include <stddef.h> + +typedef struct gpr_arena gpr_arena; + +// Create an arena, with \a initial_size bytes in the first allocated buffer +gpr_arena *gpr_arena_create(size_t initial_size); +// Allocate \a size bytes from the arena +void *gpr_arena_alloc(gpr_arena *arena, size_t size); +// Destroy an arena, returning the total number of bytes allocated +size_t gpr_arena_destroy(gpr_arena *arena); + +#endif /* GRPC_CORE_LIB_SUPPORT_ARENA_H */ diff --git a/src/core/lib/support/cmdline.c b/src/core/lib/support/cmdline.c index d47498676d..88a65a8e2e 100644 --- a/src/core/lib/support/cmdline.c +++ b/src/core/lib/support/cmdline.c @@ -71,8 +71,7 @@ struct gpr_cmdline { static int normal_state(gpr_cmdline *cl, char *arg); gpr_cmdline *gpr_cmdline_create(const char *description) { - gpr_cmdline *cl = gpr_malloc(sizeof(gpr_cmdline)); - memset(cl, 0, sizeof(gpr_cmdline)); + gpr_cmdline *cl = gpr_zalloc(sizeof(gpr_cmdline)); cl->description = description; cl->state = normal_state; @@ -101,8 +100,7 @@ static void add_arg(gpr_cmdline *cl, const char *name, const char *help, GPR_ASSERT(0 != strcmp(a->name, name)); } - a = gpr_malloc(sizeof(arg)); - memset(a, 0, sizeof(arg)); + a = gpr_zalloc(sizeof(arg)); a->name = name; a->help = help; a->type = type; diff --git a/src/core/lib/support/histogram.c b/src/core/lib/support/histogram.c index 7b016bbc78..ba8176bb05 100644 --- a/src/core/lib/support/histogram.c +++ b/src/core/lib/support/histogram.c @@ -102,8 +102,7 @@ gpr_histogram *gpr_histogram_create(double resolution, h->num_buckets = bucket_for_unchecked(h, max_bucket_start) + 1; GPR_ASSERT(h->num_buckets > 1); GPR_ASSERT(h->num_buckets < 100000000); - h->buckets = gpr_malloc(sizeof(uint32_t) * h->num_buckets); - memset(h->buckets, 0, sizeof(uint32_t) * h->num_buckets); + h->buckets = gpr_zalloc(sizeof(uint32_t) * h->num_buckets); return h; } diff --git a/src/core/lib/support/spinlock.h b/src/core/lib/support/spinlock.h new file mode 100644 index 0000000000..d8c7c5ffde --- /dev/null +++ b/src/core/lib/support/spinlock.h @@ -0,0 +1,52 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_SUPPORT_SPINLOCK_H +#define GRPC_CORE_LIB_SUPPORT_SPINLOCK_H + +#include <grpc/support/atm.h> + +/* Simple spinlock. No backoff strategy, gpr_spinlock_lock is almost always + a concurrency code smell. */ +typedef struct { gpr_atm atm; } gpr_spinlock; + +#define GPR_SPINLOCK_INITIALIZER ((gpr_spinlock){0}) +#define GPR_SPINLOCK_STATIC_INITIALIZER \ + { 0 } +#define gpr_spinlock_trylock(lock) (gpr_atm_acq_cas(&(lock)->atm, 0, 1)) +#define gpr_spinlock_unlock(lock) (gpr_atm_rel_store(&(lock)->atm, 0)) +#define gpr_spinlock_lock(lock) \ + do { \ + } while (!gpr_spinlock_trylock((lock))) + +#endif /* GRPC_CORE_LIB_SUPPORT_SPINLOCK_H */ diff --git a/src/core/lib/support/subprocess_posix.c b/src/core/lib/support/subprocess_posix.c index 4247a1c12b..ed653b9c2e 100644 --- a/src/core/lib/support/subprocess_posix.c +++ b/src/core/lib/support/subprocess_posix.c @@ -76,8 +76,7 @@ gpr_subprocess *gpr_subprocess_create(int argc, const char **argv) { _exit(1); return NULL; } else { - r = gpr_malloc(sizeof(gpr_subprocess)); - memset(r, 0, sizeof(*r)); + r = gpr_zalloc(sizeof(gpr_subprocess)); r->pid = pid; return r; } diff --git a/src/core/lib/support/sync.c b/src/core/lib/support/sync.c index 44b83f8175..b52f004f74 100644 --- a/src/core/lib/support/sync.c +++ b/src/core/lib/support/sync.c @@ -37,6 +37,8 @@ #include <grpc/support/log.h> #include <grpc/support/sync.h> +#include <assert.h> + /* Number of mutexes to allocate for events, to avoid lock contention. Should be a prime. */ enum { event_sync_partitions = 31 }; @@ -99,8 +101,12 @@ void gpr_ref_init(gpr_refcount *r, int n) { gpr_atm_rel_store(&r->count, n); } void gpr_ref(gpr_refcount *r) { gpr_atm_no_barrier_fetch_add(&r->count, 1); } void gpr_ref_non_zero(gpr_refcount *r) { +#ifndef NDEBUG gpr_atm prior = gpr_atm_no_barrier_fetch_add(&r->count, 1); - GPR_ASSERT(prior > 0); + assert(prior > 0); +#else + gpr_ref(r); +#endif } void gpr_refn(gpr_refcount *r, int n) { @@ -113,6 +119,10 @@ int gpr_unref(gpr_refcount *r) { return prior == 1; } +int gpr_ref_is_unique(gpr_refcount *r) { + return gpr_atm_acq_load(&r->count) == 1; +} + void gpr_stats_init(gpr_stats_counter *c, intptr_t n) { gpr_atm_rel_store(&c->value, n); } diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index af32c5b378..2c5d8c0ff3 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -51,6 +51,7 @@ #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" +#include "src/core/lib/support/arena.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/api_trace.h" #include "src/core/lib/surface/call.h" @@ -138,14 +139,15 @@ typedef struct batch_control { } batch_control; struct grpc_call { + gpr_arena *arena; grpc_completion_queue *cq; grpc_polling_entity pollent; grpc_channel *channel; grpc_call *parent; grpc_call *first_child; gpr_timespec start_time; - /* TODO(ctiller): share with cq if possible? */ - gpr_mu mu; + /* protects first_child, and child next/prev links */ + gpr_mu child_list_mu; /* client or server call */ bool is_client; @@ -160,7 +162,8 @@ struct grpc_call { bool received_initial_metadata; bool receiving_message; bool requested_final_op; - bool received_final_op; + gpr_atm any_ops_sent_atm; + gpr_atm received_final_op_atm; /* have we received initial metadata */ bool has_initial_md_been_received; @@ -211,6 +214,8 @@ struct grpc_call { grpc_closure receiving_initial_metadata_ready; uint32_t test_only_last_message_flags; + grpc_closure release_call; + union { struct { grpc_status_code *status; @@ -272,10 +277,13 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx, grpc_channel_get_channel_stack(args->channel); grpc_call *call; GPR_TIMER_BEGIN("grpc_call_create", 0); - call = gpr_malloc(sizeof(grpc_call) + channel_stack->call_stack_size); + gpr_arena *arena = + gpr_arena_create(grpc_channel_get_call_size_estimate(args->channel)); + call = gpr_arena_alloc(arena, + sizeof(grpc_call) + channel_stack->call_stack_size); + call->arena = arena; *out_call = call; - memset(call, 0, sizeof(grpc_call)); - gpr_mu_init(&call->mu); + gpr_mu_init(&call->child_list_mu); call->channel = args->channel; call->cq = args->cq; call->parent = args->parent_call; @@ -313,7 +321,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx, GPR_ASSERT(call->is_client); GPR_ASSERT(!args->parent_call->is_client); - gpr_mu_lock(&args->parent_call->mu); + gpr_mu_lock(&args->parent_call->child_list_mu); if (args->propagation_mask & GRPC_PROPAGATE_DEADLINE) { send_deadline = gpr_time_min( @@ -341,6 +349,10 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx, } if (args->propagation_mask & GRPC_PROPAGATE_CANCELLATION) { call->cancellation_is_inherited = 1; + if (gpr_atm_acq_load(&args->parent_call->received_final_op_atm)) { + cancel_with_error(exec_ctx, call, STATUS_FROM_API_OVERRIDE, + GRPC_ERROR_CANCELLED); + } } if (args->parent_call->first_child == NULL) { @@ -353,18 +365,23 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx, call; } - gpr_mu_unlock(&args->parent_call->mu); + gpr_mu_unlock(&args->parent_call->child_list_mu); } call->send_deadline = send_deadline; GRPC_CHANNEL_INTERNAL_REF(args->channel, "call"); /* initial refcount dropped by grpc_call_destroy */ + grpc_call_element_args call_args = { + .call_stack = CALL_STACK_FROM_CALL(call), + .server_transport_data = args->server_transport_data, + .context = call->context, + .path = path, + .start_time = call->start_time, + .deadline = send_deadline, + .arena = call->arena}; add_init_error(&error, grpc_call_stack_init(exec_ctx, channel_stack, 1, - destroy_call, call, call->context, - args->server_transport_data, path, - call->start_time, send_deadline, - CALL_STACK_FROM_CALL(call))); + destroy_call, call, &call_args)); if (error != GRPC_ERROR_NONE) { cancel_with_error(exec_ctx, call, STATUS_FROM_SURFACE, GRPC_ERROR_REF(error)); @@ -421,6 +438,14 @@ void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c REF_ARG) { GRPC_CALL_STACK_UNREF(exec_ctx, CALL_STACK_FROM_CALL(c), REF_REASON); } +static void release_call(grpc_exec_ctx *exec_ctx, void *call, + grpc_error *error) { + grpc_call *c = call; + grpc_channel *channel = c->channel; + grpc_channel_update_call_size_estimate(channel, gpr_arena_destroy(c->arena)); + GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel, "call"); +} + static void set_status_value_directly(grpc_status_code status, void *dest); static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, grpc_error *error) { @@ -435,7 +460,7 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, if (c->receiving_stream != NULL) { grpc_byte_stream_destroy(exec_ctx, c->receiving_stream); } - gpr_mu_destroy(&c->mu); + gpr_mu_destroy(&c->child_list_mu); for (ii = 0; ii < c->send_extra_metadata_count; ii++) { GRPC_MDELEM_UNREF(exec_ctx, c->send_extra_metadata[ii].md); } @@ -447,7 +472,6 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, if (c->cq) { GRPC_CQ_INTERNAL_UNREF(c->cq, "bind"); } - grpc_channel *channel = c->channel; get_final_status(call, set_status_value_directly, &c->final_info.final_status, NULL); @@ -456,11 +480,12 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, for (i = 0; i < STATUS_SOURCE_COUNT; i++) { GRPC_ERROR_UNREF( - unpack_received_status(gpr_atm_no_barrier_load(&c->status[i])).error); + unpack_received_status(gpr_atm_acq_load(&c->status[i])).error); } - grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->final_info, c); - GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel, "call"); + grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->final_info, + grpc_closure_init(&c->release_call, release_call, c, + grpc_schedule_on_exec_ctx)); GPR_TIMER_END("destroy_call", 0); } @@ -473,7 +498,7 @@ void grpc_call_destroy(grpc_call *c) { GRPC_API_TRACE("grpc_call_destroy(c=%p)", 1, (c)); if (parent) { - gpr_mu_lock(&parent->mu); + gpr_mu_lock(&parent->child_list_mu); if (c == parent->first_child) { parent->first_child = c->sibling_next; if (c == parent->first_child) { @@ -482,15 +507,14 @@ void grpc_call_destroy(grpc_call *c) { c->sibling_prev->sibling_next = c->sibling_next; c->sibling_next->sibling_prev = c->sibling_prev; } - gpr_mu_unlock(&parent->mu); + gpr_mu_unlock(&parent->child_list_mu); GRPC_CALL_INTERNAL_UNREF(&exec_ctx, parent, "child"); } - gpr_mu_lock(&c->mu); GPR_ASSERT(!c->destroy_called); c->destroy_called = 1; - cancel = !c->received_final_op; - gpr_mu_unlock(&c->mu); + cancel = gpr_atm_acq_load(&c->any_ops_sent_atm) && + !gpr_atm_acq_load(&c->received_final_op_atm); if (cancel) { cancel_with_error(&exec_ctx, c, STATUS_FROM_API_OVERRIDE, GRPC_ERROR_CANCELLED); @@ -555,53 +579,25 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c, "c=%p, status=%d, description=%s, reserved=%p)", 4, (c, (int)status, description, reserved)); GPR_ASSERT(reserved == NULL); - gpr_mu_lock(&c->mu); cancel_with_status(&exec_ctx, c, STATUS_FROM_API_OVERRIDE, status, description); - gpr_mu_unlock(&c->mu); grpc_exec_ctx_finish(&exec_ctx); return GRPC_CALL_OK; } -typedef struct termination_closure { - grpc_closure closure; - grpc_call *call; - grpc_transport_stream_op op; -} termination_closure; - -static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp, +static void done_termination(grpc_exec_ctx *exec_ctx, void *call, grpc_error *error) { - termination_closure *tc = tcp; - GRPC_CALL_INTERNAL_UNREF(exec_ctx, tc->call, "termination"); - gpr_free(tc); -} - -static void send_termination(grpc_exec_ctx *exec_ctx, void *tcp, - grpc_error *error) { - termination_closure *tc = tcp; - memset(&tc->op, 0, sizeof(tc->op)); - tc->op.cancel_error = GRPC_ERROR_REF(error); - /* reuse closure to catch completion */ - tc->op.on_complete = grpc_closure_init(&tc->closure, done_termination, tc, - grpc_schedule_on_exec_ctx); - execute_op(exec_ctx, tc->call, &tc->op); -} - -static void terminate_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c, - grpc_error *error) { - termination_closure *tc = gpr_malloc(sizeof(*tc)); - memset(tc, 0, sizeof(*tc)); - tc->call = c; - GRPC_CALL_INTERNAL_REF(tc->call, "termination"); - grpc_closure_sched(exec_ctx, grpc_closure_init(&tc->closure, send_termination, - tc, grpc_schedule_on_exec_ctx), - error); + GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "termination"); } static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c, status_source source, grpc_error *error) { + GRPC_CALL_INTERNAL_REF(c, "termination"); set_status_from_error(exec_ctx, c, source, GRPC_ERROR_REF(error)); - terminate_with_error(exec_ctx, c, error); + grpc_transport_stream_op *op = grpc_make_transport_stream_op( + grpc_closure_create(done_termination, c, grpc_schedule_on_exec_ctx)); + op->cancel_error = error; + execute_op(exec_ctx, c, op); } static grpc_error *error_from_status(grpc_status_code status, @@ -715,9 +711,7 @@ static void set_incoming_compression_algorithm( grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm( grpc_call *call) { grpc_compression_algorithm algorithm; - gpr_mu_lock(&call->mu); algorithm = call->incoming_compression_algorithm; - gpr_mu_unlock(&call->mu); return algorithm; } @@ -729,9 +723,7 @@ static grpc_compression_algorithm compression_algorithm_for_level_locked( uint32_t grpc_call_test_only_get_message_flags(grpc_call *call) { uint32_t flags; - gpr_mu_lock(&call->mu); flags = call->test_only_last_message_flags; - gpr_mu_unlock(&call->mu); return flags; } @@ -785,9 +777,7 @@ static void set_encodings_accepted_by_peer(grpc_exec_ctx *exec_ctx, uint32_t grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call) { uint32_t encodings_accepted_by_peer; - gpr_mu_lock(&call->mu); encodings_accepted_by_peer = call->encodings_accepted_by_peer; - gpr_mu_unlock(&call->mu); return encodings_accepted_by_peer; } @@ -1056,7 +1046,7 @@ static void finish_batch_completion(grpc_exec_ctx *exec_ctx, void *user_data, } static grpc_error *consolidate_batch_errors(batch_control *bctl) { - size_t n = (size_t)gpr_atm_no_barrier_load(&bctl->num_errors); + size_t n = (size_t)gpr_atm_acq_load(&bctl->num_errors); if (n == 0) { return GRPC_ERROR_NONE; } else if (n == 1) { @@ -1083,8 +1073,6 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx, grpc_call *call = bctl->call; grpc_error *error = consolidate_batch_errors(bctl); - gpr_mu_lock(&call->mu); - if (bctl->send_initial_metadata) { grpc_metadata_batch_destroy( exec_ctx, @@ -1103,20 +1091,23 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx, &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; recv_trailing_filter(exec_ctx, call, md); - call->received_final_op = true; /* propagate cancellation to any interested children */ + gpr_atm_rel_store(&call->received_final_op_atm, 1); + gpr_mu_lock(&call->child_list_mu); child_call = call->first_child; if (child_call != NULL) { do { next_child_call = child_call->sibling_next; if (child_call->cancellation_is_inherited) { GRPC_CALL_INTERNAL_REF(child_call, "propagate_cancel"); - grpc_call_cancel(child_call, NULL); + cancel_with_error(exec_ctx, child_call, STATUS_FROM_API_OVERRIDE, + GRPC_ERROR_CANCELLED); GRPC_CALL_INTERNAL_UNREF(exec_ctx, child_call, "propagate_cancel"); } child_call = next_child_call; } while (child_call != call->first_child); } + gpr_mu_unlock(&call->child_list_mu); if (call->is_client) { get_final_status(call, set_status_value_directly, @@ -1130,7 +1121,6 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx, GRPC_ERROR_UNREF(error); error = GRPC_ERROR_NONE; } - gpr_mu_unlock(&call->mu); if (bctl->is_notify_tag_closure) { /* unrefs bctl->error */ @@ -1221,7 +1211,6 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp, grpc_error *error) { batch_control *bctl = bctlp; grpc_call *call = bctl->call; - gpr_mu_lock(&bctl->call->mu); if (error != GRPC_ERROR_NONE) { if (call->receiving_stream != NULL) { grpc_byte_stream_destroy(exec_ctx, call->receiving_stream); @@ -1233,11 +1222,9 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp, } if (call->has_initial_md_been_received || error != GRPC_ERROR_NONE || call->receiving_stream == NULL) { - gpr_mu_unlock(&bctl->call->mu); process_data_after_md(exec_ctx, bctlp); } else { call->saved_receiving_stream_ready_bctlp = bctlp; - gpr_mu_unlock(&bctl->call->mu); } } @@ -1296,7 +1283,7 @@ static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx, static void add_batch_error(grpc_exec_ctx *exec_ctx, batch_control *bctl, grpc_error *error, bool has_cancelled) { if (error == GRPC_ERROR_NONE) return; - int idx = (int)gpr_atm_no_barrier_fetch_add(&bctl->num_errors, 1); + int idx = (int)gpr_atm_full_fetch_add(&bctl->num_errors, 1); if (idx == 0 && !has_cancelled) { cancel_with_error(exec_ctx, bctl->call, STATUS_FROM_CORE, GRPC_ERROR_REF(error)); @@ -1309,8 +1296,6 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx, batch_control *bctl = bctlp; grpc_call *call = bctl->call; - gpr_mu_lock(&call->mu); - add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), false); if (error == GRPC_ERROR_NONE) { grpc_metadata_batch *md = @@ -1336,11 +1321,9 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx, receiving_stream_ready, call->saved_receiving_stream_ready_bctlp, grpc_schedule_on_exec_ctx); call->saved_receiving_stream_ready_bctlp = NULL; - grpc_closure_sched(exec_ctx, saved_rsr_closure, GRPC_ERROR_REF(error)); + grpc_closure_run(exec_ctx, saved_rsr_closure, GRPC_ERROR_REF(error)); } - gpr_mu_unlock(&call->mu); - finish_batch_step(exec_ctx, bctl); } @@ -1386,7 +1369,6 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, goto done; } - /* TODO(ctiller): this feels like it could be made lock-free */ bctl = allocate_batch_control(call, ops, nops); if (bctl == NULL) { return GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; @@ -1394,7 +1376,6 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, bctl->notify_tag = notify_tag; bctl->is_notify_tag_closure = (uint8_t)(is_notify_tag_closure != 0); - gpr_mu_lock(&call->mu); grpc_transport_stream_op *stream_op = &bctl->op; memset(stream_op, 0, sizeof(*stream_op)); stream_op->covered_by_poller = true; @@ -1680,7 +1661,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, grpc_closure_init(&bctl->finish_batch, finish_batch, bctl, grpc_schedule_on_exec_ctx); stream_op->on_complete = &bctl->finish_batch; - gpr_mu_unlock(&call->mu); + gpr_atm_rel_store(&call->any_ops_sent_atm, 1); execute_op(exec_ctx, call, stream_op); @@ -1711,7 +1692,6 @@ done_with_error: if (bctl->recv_final_op) { call->requested_final_op = 0; } - gpr_mu_unlock(&call->mu); goto done; } @@ -1760,10 +1740,8 @@ uint8_t grpc_call_is_client(grpc_call *call) { return call->is_client; } grpc_compression_algorithm grpc_call_compression_for_level( grpc_call *call, grpc_compression_level level) { - gpr_mu_lock(&call->mu); grpc_compression_algorithm algo = compression_algorithm_for_level_locked(call, level); - gpr_mu_unlock(&call->mu); return algo; } diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c index d6acd392c1..2b700b2f67 100644 --- a/src/core/lib/surface/channel.c +++ b/src/core/lib/surface/channel.c @@ -68,6 +68,8 @@ struct grpc_channel { grpc_compression_options compression_options; grpc_mdelem default_authority; + gpr_atm call_size_estimate; + gpr_mu registered_call_mu; registered_call *registered_calls; @@ -115,6 +117,10 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, gpr_mu_init(&channel->registered_call_mu); channel->registered_calls = NULL; + gpr_atm_no_barrier_store( + &channel->call_size_estimate, + (gpr_atm)CHANNEL_STACK_FROM_CHANNEL(channel)->call_stack_size); + grpc_compression_options_init(&channel->compression_options); for (size_t i = 0; i < args->num_args; i++) { if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) { @@ -177,6 +183,32 @@ done: return channel; } +size_t grpc_channel_get_call_size_estimate(grpc_channel *channel) { +#define ROUND_UP_SIZE 256 + return ((size_t)gpr_atm_no_barrier_load(&channel->call_size_estimate) + + ROUND_UP_SIZE) & + ~(size_t)(ROUND_UP_SIZE - 1); +} + +void grpc_channel_update_call_size_estimate(grpc_channel *channel, + size_t size) { + size_t cur = (size_t)gpr_atm_no_barrier_load(&channel->call_size_estimate); + if (cur < size) { + /* size grew: update estimate */ + gpr_atm_no_barrier_cas(&channel->call_size_estimate, (gpr_atm)cur, + (gpr_atm)size); + /* if we lose: never mind, something else will likely update soon enough */ + } else if (cur == size) { + /* no change: holding pattern */ + } else if (cur > 0) { + /* size shrank: decrease estimate */ + gpr_atm_no_barrier_cas( + &channel->call_size_estimate, (gpr_atm)cur, + (gpr_atm)(GPR_MIN(cur - 1, (255 * cur + size) / 256))); + /* if we lose: never mind, something else will likely update soon enough */ + } +} + char *grpc_channel_get_target(grpc_channel *channel) { GRPC_API_TRACE("grpc_channel_get_target(channel=%p)", 1, (channel)); return gpr_strdup(channel->target); diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h index 3a441d7add..c4aebd8b9b 100644 --- a/src/core/lib/surface/channel.h +++ b/src/core/lib/surface/channel.h @@ -66,6 +66,9 @@ grpc_mdelem grpc_channel_get_reffed_status_elem(grpc_exec_ctx *exec_ctx, grpc_channel *channel, int status_code); +size_t grpc_channel_get_call_size_estimate(grpc_channel *channel); +void grpc_channel_update_call_size_estimate(grpc_channel *channel, size_t size); + #ifdef GRPC_STREAM_REFCOUNT_DEBUG void grpc_channel_internal_ref(grpc_channel *channel, const char *reason); void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, grpc_channel *channel, diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index b80656aa8d..b4594817e4 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -118,7 +118,7 @@ grpc_completion_queue *grpc_completion_queue_create(void *reserved) { GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved)); - cc = gpr_malloc(sizeof(grpc_completion_queue) + grpc_pollset_size()); + cc = gpr_zalloc(sizeof(grpc_completion_queue) + grpc_pollset_size()); grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu); #ifndef NDEBUG cc->outstanding_tags = NULL; diff --git a/src/core/lib/surface/lame_client.c b/src/core/lib/surface/lame_client.c index 49bc4c114b..9ddb88bd11 100644 --- a/src/core/lib/surface/lame_client.c +++ b/src/core/lib/surface/lame_client.c @@ -130,8 +130,8 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory) { - gpr_free(and_free_memory); + grpc_closure *then_schedule_closure) { + grpc_closure_sched(exec_ctx, then_schedule_closure, GRPC_ERROR_NONE); } static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index fdb230b3e2..1186a4af63 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -898,7 +898,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *ignored) { + grpc_closure *ignored) { channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; @@ -1016,12 +1016,10 @@ void grpc_server_register_non_listening_completion_queue( grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) { GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved)); - grpc_server *server = gpr_malloc(sizeof(grpc_server)); + grpc_server *server = gpr_zalloc(sizeof(grpc_server)); GPR_ASSERT(grpc_is_initialized() && "call grpc_init()"); - memset(server, 0, sizeof(grpc_server)); - gpr_mu_init(&server->mu_global); gpr_mu_init(&server->mu_call); @@ -1070,8 +1068,7 @@ void *grpc_server_register_method( flags); return NULL; } - m = gpr_malloc(sizeof(registered_method)); - memset(m, 0, sizeof(*m)); + m = gpr_zalloc(sizeof(registered_method)); m->method = gpr_strdup(method); m->host = gpr_strdup(host); m->next = server->registered_methods; @@ -1175,8 +1172,7 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, if (num_registered_methods > 0) { slots = 2 * num_registered_methods; alloc = sizeof(channel_registered_method) * slots; - chand->registered_methods = gpr_malloc(alloc); - memset(chand->registered_methods, 0, alloc); + chand->registered_methods = gpr_zalloc(alloc); for (rm = s->registered_methods; rm; rm = rm->next) { grpc_slice host; bool has_host; diff --git a/src/core/lib/surface/version.c b/src/core/lib/surface/version.c index 1143a9e044..ba80bd801e 100644 --- a/src/core/lib/surface/version.c +++ b/src/core/lib/surface/version.c @@ -38,4 +38,4 @@ const char *grpc_version_string(void) { return "3.0.0-dev"; } -const char *grpc_g_stands_for(void) { return "green"; } +const char *grpc_g_stands_for(void) { return "gentle"; } diff --git a/src/core/lib/transport/error_utils.c b/src/core/lib/transport/error_utils.c index da77828d9c..ef55e561fb 100644 --- a/src/core/lib/transport/error_utils.c +++ b/src/core/lib/transport/error_utils.c @@ -44,12 +44,12 @@ static grpc_error *recursively_find_error_with_field(grpc_error *error, } if (grpc_error_is_special(error)) return NULL; // Otherwise, search through its children. - intptr_t key = 0; - while (true) { - grpc_error *child_error = gpr_avl_get(error->errs, (void *)key++); - if (child_error == NULL) break; - grpc_error *result = recursively_find_error_with_field(child_error, which); - if (result != NULL) return result; + uint8_t slot = error->first_err; + while (slot != UINT8_MAX) { + grpc_linked_error *lerr = (grpc_linked_error *)(error->arena + slot); + grpc_error *result = recursively_find_error_with_field(lerr->err, which); + if (result) return result; + slot = lerr->next; } return NULL; } @@ -112,13 +112,13 @@ bool grpc_error_has_clear_grpc_status(grpc_error *error) { if (grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, NULL)) { return true; } - intptr_t key = 0; - while (true) { - grpc_error *child_error = gpr_avl_get(error->errs, (void *)key++); - if (child_error == NULL) break; - if (grpc_error_has_clear_grpc_status(child_error)) { + uint8_t slot = error->first_err; + while (slot != UINT8_MAX) { + grpc_linked_error *lerr = (grpc_linked_error *)(error->arena + slot); + if (grpc_error_has_clear_grpc_status(lerr->err)) { return true; } + slot = lerr->next; } return false; } diff --git a/src/core/lib/transport/metadata.c b/src/core/lib/transport/metadata.c index 489c20cbc8..f2417d8c4e 100644 --- a/src/core/lib/transport/metadata.c +++ b/src/core/lib/transport/metadata.c @@ -130,8 +130,7 @@ void grpc_mdctx_global_init(void) { shard->count = 0; gpr_atm_no_barrier_store(&shard->free_estimate, 0); shard->capacity = INITIAL_SHARD_CAPACITY; - shard->elems = gpr_malloc(sizeof(*shard->elems) * shard->capacity); - memset(shard->elems, 0, sizeof(*shard->elems) * shard->capacity); + shard->elems = gpr_zalloc(sizeof(*shard->elems) * shard->capacity); } } @@ -216,8 +215,7 @@ static void grow_mdtab(mdtab_shard *shard) { GPR_TIMER_BEGIN("grow_mdtab", 0); - mdtab = gpr_malloc(sizeof(interned_metadata *) * capacity); - memset(mdtab, 0, sizeof(interned_metadata *) * capacity); + mdtab = gpr_zalloc(sizeof(interned_metadata *) * capacity); for (i = 0; i < shard->capacity; i++) { for (md = shard->elems[i]; md; md = next) { diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c index 004e748f25..d56cb31ee0 100644 --- a/src/core/lib/transport/transport.c +++ b/src/core/lib/transport/transport.c @@ -84,6 +84,39 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, } } +#define STREAM_REF_FROM_SLICE_REF(p) \ + ((grpc_stream_refcount *)(((uint8_t *)p) - \ + offsetof(grpc_stream_refcount, slice_refcount))) + +static void slice_stream_ref(void *p) { +#ifdef GRPC_STREAM_REFCOUNT_DEBUG + grpc_stream_ref(STREAM_REF_FROM_SLICE_REF(p), "slice"); +#else + grpc_stream_ref(STREAM_REF_FROM_SLICE_REF(p)); +#endif +} + +static void slice_stream_unref(grpc_exec_ctx *exec_ctx, void *p) { +#ifdef GRPC_STREAM_REFCOUNT_DEBUG + grpc_stream_unref(exec_ctx, STREAM_REF_FROM_SLICE_REF(p), "slice"); +#else + grpc_stream_unref(exec_ctx, STREAM_REF_FROM_SLICE_REF(p)); +#endif +} + +grpc_slice grpc_slice_from_stream_owned_buffer(grpc_stream_refcount *refcount, + void *buffer, size_t length) { + slice_stream_ref(&refcount->slice_refcount); + return (grpc_slice){.refcount = &refcount->slice_refcount, + .data.refcounted = {.bytes = buffer, .length = length}}; +} + +static const grpc_slice_refcount_vtable stream_ref_slice_vtable = { + .ref = slice_stream_ref, + .unref = slice_stream_unref, + .eq = grpc_slice_default_eq_impl, + .hash = grpc_slice_default_hash_impl}; + #ifdef GRPC_STREAM_REFCOUNT_DEBUG void grpc_stream_ref_init(grpc_stream_refcount *refcount, int initial_refs, grpc_iomgr_cb_func cb, void *cb_arg, @@ -95,6 +128,8 @@ void grpc_stream_ref_init(grpc_stream_refcount *refcount, int initial_refs, #endif gpr_ref_init(&refcount->refs, initial_refs); grpc_closure_init(&refcount->destroy, cb, cb_arg, grpc_schedule_on_exec_ctx); + refcount->slice_refcount.vtable = &stream_ref_slice_vtable; + refcount->slice_refcount.sub_refcount = &refcount->slice_refcount; } static void move64(uint64_t *from, uint64_t *to) { @@ -127,9 +162,9 @@ void grpc_transport_destroy(grpc_exec_ctx *exec_ctx, int grpc_transport_init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *transport, grpc_stream *stream, grpc_stream_refcount *refcount, - const void *server_data) { + const void *server_data, gpr_arena *arena) { return transport->vtable->init_stream(exec_ctx, transport, stream, refcount, - server_data); + server_data, arena); } void grpc_transport_perform_stream_op(grpc_exec_ctx *exec_ctx, @@ -162,9 +197,10 @@ void grpc_transport_set_pops(grpc_exec_ctx *exec_ctx, grpc_transport *transport, void grpc_transport_destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *transport, - grpc_stream *stream, void *and_free_memory) { + grpc_stream *stream, + grpc_closure *then_schedule_closure) { transport->vtable->destroy_stream(exec_ctx, transport, stream, - and_free_memory); + then_schedule_closure); } char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx, @@ -219,8 +255,9 @@ typedef struct { static void destroy_made_transport_stream_op(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { made_transport_stream_op *op = arg; - grpc_closure_sched(exec_ctx, op->inner_on_complete, GRPC_ERROR_REF(error)); + grpc_closure *c = op->inner_on_complete; gpr_free(op); + grpc_closure_run(exec_ctx, c, GRPC_ERROR_REF(error)); } grpc_transport_stream_op *grpc_make_transport_stream_op( diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h index e56bf2780a..950b18aeda 100644 --- a/src/core/lib/transport/transport.h +++ b/src/core/lib/transport/transport.h @@ -41,6 +41,7 @@ #include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/iomgr/pollset.h" #include "src/core/lib/iomgr/pollset_set.h" +#include "src/core/lib/support/arena.h" #include "src/core/lib/transport/byte_stream.h" #include "src/core/lib/transport/metadata_batch.h" @@ -64,6 +65,7 @@ typedef struct grpc_stream_refcount { #ifdef GRPC_STREAM_REFCOUNT_DEBUG const char *object_type; #endif + grpc_slice_refcount slice_refcount; } grpc_stream_refcount; #ifdef GRPC_STREAM_REFCOUNT_DEBUG @@ -84,6 +86,11 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount); grpc_stream_ref_init(rc, ir, cb, cb_arg) #endif +/* Wrap a buffer that is owned by some stream object into a slice that shares + the same refcount */ +grpc_slice grpc_slice_from_stream_owned_buffer(grpc_stream_refcount *refcount, + void *buffer, size_t length); + typedef struct { uint64_t framing_bytes; uint64_t data_bytes; @@ -213,6 +220,7 @@ size_t grpc_transport_stream_size(grpc_transport *transport); /* Initialize transport data for a stream. Returns 0 on success, any other (transport-defined) value for failure. + May assume that stream contains all-zeros. Arguments: transport - the transport on which to create this stream @@ -222,7 +230,7 @@ size_t grpc_transport_stream_size(grpc_transport *transport); int grpc_transport_init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *transport, grpc_stream *stream, grpc_stream_refcount *refcount, - const void *server_data); + const void *server_data, gpr_arena *arena); void grpc_transport_set_pops(grpc_exec_ctx *exec_ctx, grpc_transport *transport, grpc_stream *stream, grpc_polling_entity *pollent); @@ -239,7 +247,8 @@ void grpc_transport_set_pops(grpc_exec_ctx *exec_ctx, grpc_transport *transport, caller, but any child memory must be cleaned up) */ void grpc_transport_destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *transport, - grpc_stream *stream, void *and_free_memory); + grpc_stream *stream, + grpc_closure *then_schedule_closure); void grpc_transport_stream_op_finish_with_failure(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op, diff --git a/src/core/lib/transport/transport_impl.h b/src/core/lib/transport/transport_impl.h index 8553148c35..6f688bf8d2 100644 --- a/src/core/lib/transport/transport_impl.h +++ b/src/core/lib/transport/transport_impl.h @@ -47,7 +47,7 @@ typedef struct grpc_transport_vtable { /* implementation of grpc_transport_init_stream */ int (*init_stream)(grpc_exec_ctx *exec_ctx, grpc_transport *self, grpc_stream *stream, grpc_stream_refcount *refcount, - const void *server_data); + const void *server_data, gpr_arena *arena); /* implementation of grpc_transport_set_pollset */ void (*set_pollset)(grpc_exec_ctx *exec_ctx, grpc_transport *self, @@ -67,7 +67,8 @@ typedef struct grpc_transport_vtable { /* implementation of grpc_transport_destroy_stream */ void (*destroy_stream)(grpc_exec_ctx *exec_ctx, grpc_transport *self, - grpc_stream *stream, void *and_free_memory); + grpc_stream *stream, + grpc_closure *then_schedule_closure); /* implementation of grpc_transport_destroy */ void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_transport *self); diff --git a/src/core/lib/tsi/fake_transport_security.c b/src/core/lib/tsi/fake_transport_security.c index 0e20d6fd71..bbe323df3b 100644 --- a/src/core/lib/tsi/fake_transport_security.c +++ b/src/core/lib/tsi/fake_transport_security.c @@ -502,8 +502,7 @@ static const tsi_handshaker_vtable handshaker_vtable = { }; tsi_handshaker *tsi_create_fake_handshaker(int is_client) { - tsi_fake_handshaker *impl = gpr_malloc(sizeof(*impl)); - memset(impl, 0, sizeof(*impl)); + tsi_fake_handshaker *impl = gpr_zalloc(sizeof(*impl)); impl->base.vtable = &handshaker_vtable; impl->is_client = is_client; impl->result = TSI_HANDSHAKE_IN_PROGRESS; @@ -519,8 +518,7 @@ tsi_handshaker *tsi_create_fake_handshaker(int is_client) { tsi_frame_protector *tsi_create_fake_protector( size_t *max_protected_frame_size) { - tsi_fake_frame_protector *impl = gpr_malloc(sizeof(*impl)); - memset(impl, 0, sizeof(*impl)); + tsi_fake_frame_protector *impl = gpr_zalloc(sizeof(*impl)); impl->max_frame_size = (max_protected_frame_size == NULL) ? TSI_FAKE_DEFAULT_FRAME_SIZE : *max_protected_frame_size; diff --git a/src/core/lib/tsi/ssl_transport_security.c b/src/core/lib/tsi/ssl_transport_security.c index 366dca9507..53aabdb926 100644 --- a/src/core/lib/tsi/ssl_transport_security.c +++ b/src/core/lib/tsi/ssl_transport_security.c @@ -976,9 +976,7 @@ static tsi_result ssl_handshaker_extract_peer(tsi_handshaker *self, if (alpn_selected != NULL) { size_t i; tsi_peer_property *new_properties = - gpr_malloc(sizeof(*new_properties) * (peer->property_count + 1)); - memset(new_properties, 0, - sizeof(*new_properties) * (peer->property_count + 1)); + gpr_zalloc(sizeof(*new_properties) * (peer->property_count + 1)); for (i = 0; i < peer->property_count; i++) { new_properties[i] = peer->properties[i]; } @@ -1002,8 +1000,7 @@ static tsi_result ssl_handshaker_create_frame_protector( size_t actual_max_output_protected_frame_size = TSI_SSL_MAX_PROTECTED_FRAME_SIZE_UPPER_BOUND; tsi_ssl_handshaker *impl = (tsi_ssl_handshaker *)self; - tsi_ssl_frame_protector *protector_impl = gpr_malloc(sizeof(*protector_impl)); - memset(protector_impl, 0, sizeof(*protector_impl)); + tsi_ssl_frame_protector *protector_impl = gpr_zalloc(sizeof(*protector_impl)); if (max_output_protected_frame_size != NULL) { if (*max_output_protected_frame_size > @@ -1119,8 +1116,7 @@ static tsi_result create_tsi_ssl_handshaker(SSL_CTX *ctx, int is_client, SSL_set_accept_state(ssl); } - impl = gpr_malloc(sizeof(*impl)); - memset(impl, 0, sizeof(*impl)); + impl = gpr_zalloc(sizeof(*impl)); impl->ssl = ssl; impl->into_ssl = into_ssl; impl->from_ssl = from_ssl; @@ -1338,8 +1334,7 @@ tsi_result tsi_create_ssl_client_handshaker_factory( return TSI_INVALID_ARGUMENT; } - impl = gpr_malloc(sizeof(*impl)); - memset(impl, 0, sizeof(*impl)); + impl = gpr_zalloc(sizeof(*impl)); impl->ssl_context = ssl_context; do { @@ -1433,17 +1428,13 @@ tsi_result tsi_create_ssl_server_handshaker_factory_ex( return TSI_INVALID_ARGUMENT; } - impl = gpr_malloc(sizeof(*impl)); - memset(impl, 0, sizeof(*impl)); + impl = gpr_zalloc(sizeof(*impl)); impl->base.create_handshaker = ssl_server_handshaker_factory_create_handshaker; impl->base.destroy = ssl_server_handshaker_factory_destroy; - impl->ssl_contexts = gpr_malloc(key_cert_pair_count * sizeof(SSL_CTX *)); - memset(impl->ssl_contexts, 0, key_cert_pair_count * sizeof(SSL_CTX *)); + impl->ssl_contexts = gpr_zalloc(key_cert_pair_count * sizeof(SSL_CTX *)); impl->ssl_context_x509_subject_names = - gpr_malloc(key_cert_pair_count * sizeof(tsi_peer)); - memset(impl->ssl_context_x509_subject_names, 0, - key_cert_pair_count * sizeof(tsi_peer)); + gpr_zalloc(key_cert_pair_count * sizeof(tsi_peer)); if (impl->ssl_contexts == NULL || impl->ssl_context_x509_subject_names == NULL) { tsi_ssl_handshaker_factory_destroy(&impl->base); diff --git a/src/core/lib/tsi/test_creds/BUILD b/src/core/lib/tsi/test_creds/BUILD new file mode 100644 index 0000000000..5cf04caf17 --- /dev/null +++ b/src/core/lib/tsi/test_creds/BUILD @@ -0,0 +1,36 @@ +# Copyright 2017, Google Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following disclaimer +# in the documentation and/or other materials provided with the +# distribution. +# * Neither the name of Google Inc. nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +licenses(["notice"]) # 3-clause BSD + +exports_files([ + "ca.pem", + "server1.key", + "server1.pem", +]) diff --git a/src/core/lib/tsi/transport_security.c b/src/core/lib/tsi/transport_security.c index 830cf09584..2cbf381c88 100644 --- a/src/core/lib/tsi/transport_security.c +++ b/src/core/lib/tsi/transport_security.c @@ -231,8 +231,7 @@ tsi_result tsi_construct_allocated_string_peer_property( *property = tsi_init_peer_property(); if (name != NULL) property->name = gpr_strdup(name); if (value_length > 0) { - property->value.data = gpr_malloc(value_length); - memset(property->value.data, 0, value_length); + property->value.data = gpr_zalloc(value_length); property->value.length = value_length; } return TSI_OK; @@ -260,8 +259,7 @@ tsi_result tsi_construct_string_peer_property(const char *name, tsi_result tsi_construct_peer(size_t property_count, tsi_peer *peer) { memset(peer, 0, sizeof(tsi_peer)); if (property_count > 0) { - peer->properties = gpr_malloc(property_count * sizeof(tsi_peer_property)); - memset(peer->properties, 0, property_count * sizeof(tsi_peer_property)); + peer->properties = gpr_zalloc(property_count * sizeof(tsi_peer_property)); peer->property_count = property_count; } return TSI_OK; |