diff options
Diffstat (limited to 'src')
29 files changed, 582 insertions, 352 deletions
diff --git a/src/compiler/python_generator.cc b/src/compiler/python_generator.cc index 4841da8da8..242ce06a16 100644 --- a/src/compiler/python_generator.cc +++ b/src/compiler/python_generator.cc @@ -101,18 +101,20 @@ class IndentScope { // TODO(https://github.com/google/protobuf/issues/888): // Export `ModuleName` from protobuf's // `src/google/protobuf/compiler/python/python_generator.cc` file. -grpc::string ModuleName(const grpc::string& filename) { +grpc::string ModuleName(const grpc::string& filename, + const grpc::string& import_prefix) { grpc::string basename = StripProto(filename); basename = StringReplace(basename, "-", "_"); basename = StringReplace(basename, "/", "."); - return basename + "_pb2"; + return import_prefix + basename + "_pb2"; } // TODO(https://github.com/google/protobuf/issues/888): // Export `ModuleAlias` from protobuf's // `src/google/protobuf/compiler/python/python_generator.cc` file. -grpc::string ModuleAlias(const grpc::string& filename) { - grpc::string module_name = ModuleName(filename); +grpc::string ModuleAlias(const grpc::string& filename, + const grpc::string& import_prefix) { + grpc::string module_name = ModuleName(filename, import_prefix); // We can't have dots in the module name, so we replace each with _dot_. // But that could lead to a collision between a.b and a_dot_b, so we also // duplicate each underscore. @@ -189,7 +191,7 @@ bool PrivateGenerator::GetModuleAndMessagePath(const Descriptor* type, grpc::string generator_file_name = file->name(); grpc::string module; if (generator_file_name != file_name || generate_in_pb2_grpc) { - module = ModuleAlias(file_name) + "."; + module = ModuleAlias(file_name, config.import_prefix) + "."; } else { module = ""; } @@ -666,8 +668,10 @@ bool PrivateGenerator::PrintPreamble() { for (int k = 0; k < 2; ++k) { const Descriptor* type = types[k]; grpc::string type_file_name = type->file()->name(); - grpc::string module_name = ModuleName(type_file_name); - grpc::string module_alias = ModuleAlias(type_file_name); + grpc::string module_name = + ModuleName(type_file_name, config.import_prefix); + grpc::string module_alias = + ModuleAlias(type_file_name, config.import_prefix); imports_set.insert(std::make_tuple(module_name, module_alias)); } } @@ -766,7 +770,9 @@ pair<bool, grpc::string> PrivateGenerator::GetGrpcServices() { } // namespace GeneratorConfiguration::GeneratorConfiguration() - : grpc_package_root("grpc"), beta_package_root("grpc.beta") {} + : grpc_package_root("grpc"), + beta_package_root("grpc.beta"), + import_prefix("") {} PythonGrpcGenerator::PythonGrpcGenerator(const GeneratorConfiguration& config) : config_(config) {} diff --git a/src/compiler/python_generator.h b/src/compiler/python_generator.h index 6a95255d40..b91059fad7 100644 --- a/src/compiler/python_generator.h +++ b/src/compiler/python_generator.h @@ -45,7 +45,10 @@ namespace grpc_python_generator { struct GeneratorConfiguration { GeneratorConfiguration(); grpc::string grpc_package_root; + // TODO(https://github.com/grpc/grpc/issues/8622): Drop this. grpc::string beta_package_root; + // TODO(https://github.com/google/protobuf/issues/888): Drop this. + grpc::string import_prefix; }; class PythonGrpcGenerator : public grpc::protobuf::compiler::CodeGenerator { diff --git a/src/core/ext/census/grpc_filter.c b/src/core/ext/census/grpc_filter.c index b80d831557..69d488ad18 100644 --- a/src/core/ext/census/grpc_filter.c +++ b/src/core/ext/census/grpc_filter.c @@ -78,7 +78,8 @@ static void client_mutate_op(grpc_call_element *elem, call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; if (op->send_initial_metadata) { - extract_and_annotate_method_tag(op->send_initial_metadata, calld, chand); + extract_and_annotate_method_tag( + op->payload->send_initial_metadata.send_initial_metadata, calld, chand); } } @@ -107,9 +108,12 @@ static void server_mutate_op(grpc_call_element *elem, call_data *calld = elem->call_data; if (op->recv_initial_metadata) { /* substitute our callback for the op callback */ - calld->recv_initial_metadata = op->recv_initial_metadata; - calld->on_done_recv = op->recv_initial_metadata_ready; - op->recv_initial_metadata_ready = &calld->finish_recv; + calld->recv_initial_metadata = + op->payload->recv_initial_metadata.recv_initial_metadata; + calld->on_done_recv = + op->payload->recv_initial_metadata.recv_initial_metadata_ready; + op->payload->recv_initial_metadata.recv_initial_metadata_ready = + &calld->finish_recv; } } diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c index d308096212..3d9d70559a 100644 --- a/src/core/ext/client_channel/client_channel.c +++ b/src/core/ext/client_channel/client_channel.c @@ -71,7 +71,8 @@ */ typedef enum { - WAIT_FOR_READY_UNSET, + /* zero so it can be default initialized */ + WAIT_FOR_READY_UNSET = 0, WAIT_FOR_READY_FALSE, WAIT_FOR_READY_TRUE } wait_for_ready_value; @@ -631,7 +632,8 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx, #define CANCELLED_CALL ((grpc_subchannel_call *)1) typedef enum { - GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING, + /* zero so that it can be default-initialized */ + GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING = 0, GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL } subchannel_creation_phase; @@ -653,7 +655,6 @@ typedef struct client_channel_call_data { gpr_timespec call_start_time; gpr_timespec deadline; method_parameters *method_params; - grpc_closure read_service_config; grpc_error *cancel_error; @@ -726,6 +727,47 @@ static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) { gpr_free(ops); } +// Sets calld->method_params. +// If the method params specify a timeout, populates +// *per_method_deadline and returns true. +static bool set_call_method_params_from_service_config_locked( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + gpr_timespec *per_method_deadline) { + channel_data *chand = elem->channel_data; + call_data *calld = elem->call_data; + if (chand->method_params_table != NULL) { + calld->method_params = grpc_method_config_table_get( + exec_ctx, chand->method_params_table, calld->path); + if (calld->method_params != NULL) { + method_parameters_ref(calld->method_params); + if (gpr_time_cmp(calld->method_params->timeout, + gpr_time_0(GPR_TIMESPAN)) != 0) { + *per_method_deadline = + gpr_time_add(calld->call_start_time, calld->method_params->timeout); + return true; + } + } + } + return false; +} + +static void apply_final_configuration_locked(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem) { + /* apply service-config level configuration to the call (now that we're + * certain it exists) */ + call_data *calld = elem->call_data; + gpr_timespec per_method_deadline; + if (set_call_method_params_from_service_config_locked(exec_ctx, elem, + &per_method_deadline)) { + // If the deadline from the service config is shorter than the one + // from the client API, reset the deadline timer. + if (gpr_time_cmp(per_method_deadline, calld->deadline) < 0) { + calld->deadline = per_method_deadline; + grpc_deadline_state_reset(exec_ctx, elem, calld->deadline); + } + } +} + static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_call_element *elem = arg; @@ -851,6 +893,7 @@ static bool pick_subchannel_locked( } GPR_ASSERT(error == GRPC_ERROR_NONE); if (chand->lb_policy != NULL) { + apply_final_configuration_locked(exec_ctx, elem); grpc_lb_policy *lb_policy = chand->lb_policy; GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel"); // If the application explicitly set wait_for_ready, use that. @@ -1063,114 +1106,18 @@ static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx, GPR_TIMER_END("cc_start_transport_stream_op", 0); } -// Sets calld->method_params. -// If the method params specify a timeout, populates -// *per_method_deadline and returns true. -static bool set_call_method_params_from_service_config_locked( - grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - gpr_timespec *per_method_deadline) { - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; - if (chand->method_params_table != NULL) { - calld->method_params = grpc_method_config_table_get( - exec_ctx, chand->method_params_table, calld->path); - if (calld->method_params != NULL) { - method_parameters_ref(calld->method_params); - if (gpr_time_cmp(calld->method_params->timeout, - gpr_time_0(GPR_TIMESPAN)) != 0) { - *per_method_deadline = - gpr_time_add(calld->call_start_time, calld->method_params->timeout); - return true; - } - } - } - return false; -} - -// Gets data from the service config. Invoked when the resolver returns -// its initial result. -static void read_service_config_locked(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { - grpc_call_element *elem = arg; - call_data *calld = elem->call_data; - // If this is an error, there's no point in looking at the service config. - if (error == GRPC_ERROR_NONE) { - gpr_timespec per_method_deadline; - if (set_call_method_params_from_service_config_locked( - exec_ctx, elem, &per_method_deadline)) { - // If the deadline from the service config is shorter than the one - // from the client API, reset the deadline timer. - if (gpr_time_cmp(per_method_deadline, calld->deadline) < 0) { - calld->deadline = per_method_deadline; - grpc_deadline_state_reset(exec_ctx, elem, calld->deadline); - } - } - } - GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config"); -} - -static void initial_read_service_config_locked(grpc_exec_ctx *exec_ctx, - void *arg, - grpc_error *error_ignored) { - grpc_call_element *elem = arg; - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; - // If the resolver has already returned results, then we can access - // the service config parameters immediately. Otherwise, we need to - // defer that work until the resolver returns an initial result. - if (chand->lb_policy != NULL) { - // We already have a resolver result, so check for service config. - gpr_timespec per_method_deadline; - if (set_call_method_params_from_service_config_locked( - exec_ctx, elem, &per_method_deadline)) { - calld->deadline = gpr_time_min(calld->deadline, per_method_deadline); - } - } else { - // We don't yet have a resolver result, so register a callback to - // get the service config data once the resolver returns. - // Take a reference to the call stack to be owned by the callback. - GRPC_CALL_STACK_REF(calld->owning_call, "read_service_config"); - grpc_closure_init(&calld->read_service_config, read_service_config_locked, - elem, grpc_combiner_scheduler(chand->combiner, false)); - grpc_closure_list_append(&chand->waiting_for_config_closures, - &calld->read_service_config, GRPC_ERROR_NONE); - } - // Start the deadline timer with the current deadline value. If we - // do not yet have service config data, then the timer may be reset - // later. - grpc_deadline_state_start(exec_ctx, elem, calld->deadline); - GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, - "initial_read_service_config"); -} - /* Constructor for call_data */ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_element_args *args) { - channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; // Initialize data members. grpc_deadline_state_init(exec_ctx, elem, args->call_stack); calld->path = grpc_slice_ref_internal(args->path); calld->call_start_time = args->start_time; calld->deadline = gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC); - calld->method_params = NULL; - calld->cancel_error = GRPC_ERROR_NONE; - gpr_atm_rel_store(&calld->subchannel_call, 0); - calld->connected_subchannel = NULL; - calld->waiting_ops = NULL; - calld->waiting_ops_count = 0; - calld->waiting_ops_capacity = 0; - calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; calld->owning_call = args->call_stack; - calld->pollent = NULL; - GRPC_CALL_STACK_REF(calld->owning_call, "initial_read_service_config"); - grpc_closure_sched( - exec_ctx, - grpc_closure_init(&calld->read_service_config, - initial_read_service_config_locked, elem, - grpc_combiner_scheduler(chand->combiner, false)), - GRPC_ERROR_NONE); + grpc_deadline_state_start(exec_ctx, elem, calld->deadline); return GRPC_ERROR_NONE; } diff --git a/src/core/ext/client_channel/parse_address.c b/src/core/ext/client_channel/parse_address.c index 8ae15fc72b..cd1b2cd80c 100644 --- a/src/core/ext/client_channel/parse_address.c +++ b/src/core/ext/client_channel/parse_address.c @@ -128,6 +128,7 @@ int parse_ipv6(grpc_uri *uri, grpc_resolved_address *resolved_addr) { GPR_ASSERT(host_end >= host); char host_without_scope[INET6_ADDRSTRLEN]; size_t host_without_scope_len = (size_t)(host_end - host); + uint32_t sin6_scope_id = 0; strncpy(host_without_scope, host, host_without_scope_len); host_without_scope[host_without_scope_len] = '\0'; if (inet_pton(AF_INET6, host_without_scope, &in6->sin6_addr) == 0) { @@ -136,10 +137,12 @@ int parse_ipv6(grpc_uri *uri, grpc_resolved_address *resolved_addr) { } if (gpr_parse_bytes_to_uint32(host_end + 1, strlen(host) - host_without_scope_len - 1, - &in6->sin6_scope_id) == 0) { + &sin6_scope_id) == 0) { gpr_log(GPR_ERROR, "invalid ipv6 scope id: '%s'", host_end + 1); goto done; } + // Handle "sin6_scope_id" being type "u_long". See grpc issue ##10027. + in6->sin6_scope_id = sin6_scope_id; } else { if (inet_pton(AF_INET6, host, &in6->sin6_addr) == 0) { gpr_log(GPR_ERROR, "invalid ipv6 address: '%s'", host); diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index de76ed4743..8e403086c0 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -511,6 +511,10 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_error *error) { if (!t->closed) { + if (!grpc_error_has_clear_grpc_status(error)) { + error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, + GRPC_STATUS_UNAVAILABLE); + } if (t->write_state != GRPC_CHTTP2_WRITE_STATE_IDLE) { if (t->close_transport_on_writes_finished == NULL) { t->close_transport_on_writes_finished = @@ -520,10 +524,6 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, grpc_error_add_child(t->close_transport_on_writes_finished, error); return; } - if (!grpc_error_has_clear_grpc_status(error)) { - error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_UNAVAILABLE); - } t->closed = 1; connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "close_transport"); diff --git a/src/core/ext/transport/chttp2/transport/frame_ping.c b/src/core/ext/transport/chttp2/transport/frame_ping.c index f487533c41..9b4b1a7b84 100644 --- a/src/core/ext/transport/chttp2/transport/frame_ping.c +++ b/src/core/ext/transport/chttp2/transport/frame_ping.c @@ -91,7 +91,7 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, grpc_chttp2_ping_parser *p = parser; while (p->byte != 8 && cur != end) { - p->opaque_8bytes |= (((uint64_t)*cur) << (8 * p->byte)); + p->opaque_8bytes |= (((uint64_t)*cur) << (56 - 8 * p->byte)); cur++; p->byte++; } diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c index 01a03533da..cf5acfaac3 100644 --- a/src/core/ext/transport/cronet/transport/cronet_transport.c +++ b/src/core/ext/transport/cronet/transport/cronet_transport.c @@ -54,6 +54,7 @@ #include "third_party/objective_c/Cronet/bidirectional_stream_c.h" #define GRPC_HEADER_SIZE_IN_BYTES 5 +#define GRPC_FLUSH_READ_SIZE 4096 #define CRONET_LOG(...) \ do { \ @@ -151,11 +152,17 @@ struct write_state { struct op_state { bool state_op_done[OP_NUM_OPS]; bool state_callback_received[OP_NUM_OPS]; + /* A non-zero gRPC status code has been seen */ bool fail_state; + /* Transport is discarding all buffered messages */ bool flush_read; bool flush_cronet_when_ready; bool pending_write_for_trailer; - bool unprocessed_send_message; + bool pending_send_message; + /* User requested RECV_TRAILING_METADATA */ + bool pending_recv_trailing_metadata; + /* Cronet has not issued a callback of a bidirectional read */ + bool pending_read_from_cronet; grpc_error *cancel_error; /* data structure for storing data coming from server */ struct read_state rs; @@ -248,11 +255,35 @@ static const char *op_id_string(enum e_op_id i) { return "UNKNOWN"; } -static void free_read_buffer(stream_obj *s) { +static void null_and_maybe_free_read_buffer(stream_obj *s) { if (s->state.rs.read_buffer && s->state.rs.read_buffer != s->state.rs.grpc_header_bytes) { gpr_free(s->state.rs.read_buffer); - s->state.rs.read_buffer = NULL; + } + s->state.rs.read_buffer = NULL; +} + +static void maybe_flush_read(stream_obj *s) { + /* To enter flush read state (discarding all the buffered messages in + * transport layer), two conditions must be satisfied: 1) non-zero grpc status + * has been received, and 2) an op requesting the status code + * (RECV_TRAILING_METADATA) is issued by the user. (See + * doc/status_ordering.md) */ + /* Whenever the evaluation of any of the two condition is changed, we check + * whether we should enter the flush read state. */ + if (s->state.pending_recv_trailing_metadata && s->state.fail_state) { + if (!s->state.flush_read) { + CRONET_LOG(GPR_DEBUG, "%p: Flush read", s); + s->state.flush_read = true; + null_and_maybe_free_read_buffer(s); + s->state.rs.read_buffer = gpr_malloc(GRPC_FLUSH_READ_SIZE); + if (!s->state.pending_read_from_cronet) { + CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); + bidirectional_stream_read(s->cbs, s->state.rs.read_buffer, + GRPC_FLUSH_READ_SIZE); + s->state.pending_read_from_cronet = true; + } + } } } @@ -279,7 +310,11 @@ static void add_to_storage(struct stream_obj *s, grpc_transport_stream_op *op) { storage->head = new_op; storage->num_pending_ops++; if (op->send_message) { - s->state.unprocessed_send_message = true; + s->state.pending_send_message = true; + } + if (op->recv_trailing_metadata) { + s->state.pending_recv_trailing_metadata = true; + maybe_flush_read(s); } CRONET_LOG(GPR_DEBUG, "adding new op %p. %d in the queue.", new_op, storage->num_pending_ops); @@ -367,7 +402,7 @@ static void on_failed(bidirectional_stream *stream, int net_error) { gpr_free(s->state.ws.write_buffer); s->state.ws.write_buffer = NULL; } - free_read_buffer(s); + null_and_maybe_free_read_buffer(s); gpr_mu_unlock(&s->mu); execute_from_storage(s); } @@ -390,7 +425,7 @@ static void on_canceled(bidirectional_stream *stream) { gpr_free(s->state.ws.write_buffer); s->state.ws.write_buffer = NULL; } - free_read_buffer(s); + null_and_maybe_free_read_buffer(s); gpr_mu_unlock(&s->mu); execute_from_storage(s); } @@ -405,7 +440,7 @@ static void on_succeeded(bidirectional_stream *stream) { bidirectional_stream_destroy(s->cbs); s->state.state_callback_received[OP_SUCCEEDED] = true; s->cbs = NULL; - free_read_buffer(s); + null_and_maybe_free_read_buffer(s); gpr_mu_unlock(&s->mu); execute_from_storage(s); } @@ -473,6 +508,7 @@ static void on_response_headers_received( CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); bidirectional_stream_read(s->cbs, s->state.rs.read_buffer, s->state.rs.remaining_bytes); + s->state.pending_read_from_cronet = true; } gpr_mu_unlock(&s->mu); grpc_exec_ctx_finish(&exec_ctx); @@ -504,10 +540,13 @@ static void on_read_completed(bidirectional_stream *stream, char *data, CRONET_LOG(GPR_DEBUG, "R: on_read_completed(%p, %p, %d)", stream, data, count); gpr_mu_lock(&s->mu); + s->state.pending_read_from_cronet = false; s->state.state_callback_received[OP_RECV_MESSAGE] = true; if (count > 0 && s->state.flush_read) { CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); - bidirectional_stream_read(s->cbs, s->state.rs.read_buffer, 4096); + bidirectional_stream_read(s->cbs, s->state.rs.read_buffer, + GRPC_FLUSH_READ_SIZE); + s->state.pending_read_from_cronet = true; gpr_mu_unlock(&s->mu); } else if (count > 0) { s->state.rs.received_bytes += count; @@ -518,16 +557,14 @@ static void on_read_completed(bidirectional_stream *stream, char *data, bidirectional_stream_read( s->cbs, s->state.rs.read_buffer + s->state.rs.received_bytes, s->state.rs.remaining_bytes); + s->state.pending_read_from_cronet = true; gpr_mu_unlock(&s->mu); } else { gpr_mu_unlock(&s->mu); execute_from_storage(s); } } else { - if (s->state.flush_read) { - gpr_free(s->state.rs.read_buffer); - s->state.rs.read_buffer = NULL; - } + null_and_maybe_free_read_buffer(s); s->state.rs.read_stream_closed = true; gpr_mu_unlock(&s->mu); execute_from_storage(s); @@ -564,6 +601,7 @@ static void on_response_trailers_received( if (0 == strcmp(trailers->headers[i].key, "grpc-status") && 0 != strcmp(trailers->headers[i].value, "0")) { s->state.fail_state = true; + maybe_flush_read(s); } } s->state.state_callback_received[OP_RECV_TRAILING_METADATA] = true; @@ -778,7 +816,7 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op, else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) result = false; /* we haven't sent message yet */ - else if (stream_state->unprocessed_send_message && + else if (stream_state->pending_send_message && !stream_state->state_op_done[OP_SEND_MESSAGE]) result = false; /* we haven't got on_write_completed for the send yet */ @@ -878,9 +916,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, char *url = NULL; const char *method = "POST"; s->header_array.headers = NULL; - convert_metadata_to_cronet_headers( - stream_op->send_initial_metadata->list.head, t->host, &url, - &s->header_array.headers, &s->header_array.count, &method); + convert_metadata_to_cronet_headers(stream_op->payload->send_initial_metadata + .send_initial_metadata->list.head, + t->host, &url, &s->header_array.headers, + &s->header_array.count, &method); s->header_array.capacity = s->header_array.count; CRONET_LOG(GPR_DEBUG, "bidirectional_stream_start(%p, %s)", s->cbs, url); bidirectional_stream_start(s->cbs, url, 0, method, &s->header_array, false); @@ -900,7 +939,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, } else if (stream_op->send_message && op_can_be_run(stream_op, s, &oas->state, OP_SEND_MESSAGE)) { CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_MESSAGE", oas); - stream_state->unprocessed_send_message = false; + stream_state->pending_send_message = false; if (stream_state->state_callback_received[OP_FAILED]) { result = NO_ACTION_POSSIBLE; CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed."); @@ -908,13 +947,14 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, grpc_slice_buffer write_slice_buffer; grpc_slice slice; grpc_slice_buffer_init(&write_slice_buffer); - grpc_byte_stream_next(NULL, stream_op->send_message, &slice, - stream_op->send_message->length, NULL); + grpc_byte_stream_next( + NULL, stream_op->payload->send_message.send_message, &slice, + stream_op->payload->send_message.send_message->length, NULL); /* Check that compression flag is OFF. We don't support compression yet. */ - if (stream_op->send_message->flags != 0) { + if (stream_op->payload->send_message.send_message->flags != 0) { gpr_log(GPR_ERROR, "Compression is not supported"); - GPR_ASSERT(stream_op->send_message->flags == 0); + GPR_ASSERT(stream_op->payload->send_message.send_message->flags == 0); } grpc_slice_buffer_add(&write_slice_buffer, slice); if (write_slice_buffer.count != 1) { @@ -972,17 +1012,23 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, OP_RECV_INITIAL_METADATA)) { CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_INITIAL_METADATA", oas); if (stream_state->state_op_done[OP_CANCEL_ERROR]) { - grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready, - GRPC_ERROR_NONE); + grpc_closure_sched( + exec_ctx, + stream_op->payload->recv_initial_metadata.recv_initial_metadata_ready, + GRPC_ERROR_NONE); } else if (stream_state->state_callback_received[OP_FAILED]) { - grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready, - GRPC_ERROR_NONE); + grpc_closure_sched( + exec_ctx, + stream_op->payload->recv_initial_metadata.recv_initial_metadata_ready, + GRPC_ERROR_NONE); } else { grpc_chttp2_incoming_metadata_buffer_publish( exec_ctx, &oas->s->state.rs.initial_metadata, - stream_op->recv_initial_metadata); - grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready, - GRPC_ERROR_NONE); + stream_op->payload->recv_initial_metadata.recv_initial_metadata); + grpc_closure_sched( + exec_ctx, + stream_op->payload->recv_initial_metadata.recv_initial_metadata_ready, + GRPC_ERROR_NONE); } stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true; result = ACTION_TAKEN_NO_CALLBACK; @@ -991,20 +1037,31 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_MESSAGE", oas); if (stream_state->state_op_done[OP_CANCEL_ERROR]) { CRONET_LOG(GPR_DEBUG, "Stream is cancelled."); - grpc_closure_sched(exec_ctx, stream_op->recv_message_ready, + grpc_closure_sched(exec_ctx, + stream_op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE); stream_state->state_op_done[OP_RECV_MESSAGE] = true; result = ACTION_TAKEN_NO_CALLBACK; } else if (stream_state->state_callback_received[OP_FAILED]) { CRONET_LOG(GPR_DEBUG, "Stream failed."); - grpc_closure_sched(exec_ctx, stream_op->recv_message_ready, + grpc_closure_sched(exec_ctx, + stream_op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE); stream_state->state_op_done[OP_RECV_MESSAGE] = true; result = ACTION_TAKEN_NO_CALLBACK; } else if (stream_state->rs.read_stream_closed == true) { /* No more data will be received */ CRONET_LOG(GPR_DEBUG, "read stream closed"); - grpc_closure_sched(exec_ctx, stream_op->recv_message_ready, + grpc_closure_sched(exec_ctx, + stream_op->payload->recv_message.recv_message_ready, + GRPC_ERROR_NONE); + stream_state->state_op_done[OP_RECV_MESSAGE] = true; + oas->state.state_op_done[OP_RECV_MESSAGE] = true; + result = ACTION_TAKEN_NO_CALLBACK; + } else if (stream_state->flush_read) { + CRONET_LOG(GPR_DEBUG, "flush read"); + grpc_closure_sched(exec_ctx, + stream_op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE); stream_state->state_op_done[OP_RECV_MESSAGE] = true; oas->state.state_op_done[OP_RECV_MESSAGE] = true; @@ -1029,6 +1086,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, true; /* Indicates that at least one read request has been made */ bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, stream_state->rs.remaining_bytes); + stream_state->pending_read_from_cronet = true; result = ACTION_TAKEN_WITH_CALLBACK; } else { stream_state->rs.remaining_bytes = 0; @@ -1038,8 +1096,9 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, &stream_state->rs.read_slice_buffer, 0); *((grpc_byte_buffer **)stream_op->recv_message) = (grpc_byte_buffer *)&stream_state->rs.sbs; - grpc_closure_sched(exec_ctx, stream_op->recv_message_ready, - GRPC_ERROR_NONE); + grpc_closure_sched( + exec_ctx, stream_op->payload->recv_message.recv_message_ready, + GRPC_ERROR_NONE); stream_state->state_op_done[OP_RECV_MESSAGE] = true; oas->state.state_op_done[OP_RECV_MESSAGE] = true; @@ -1047,11 +1106,13 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes; stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES; stream_state->rs.received_bytes = 0; + stream_state->rs.length_field_received = false; CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); stream_state->state_op_done[OP_READ_REQ_MADE] = true; /* Indicates that at least one read request has been made */ bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, stream_state->rs.remaining_bytes); + stream_state->pending_read_from_cronet = true; result = ACTION_TAKEN_NO_CALLBACK; } } else if (stream_state->rs.remaining_bytes == 0) { @@ -1064,6 +1125,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, true; /* Indicates that at least one read request has been made */ bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, stream_state->rs.remaining_bytes); + stream_state->pending_read_from_cronet = true; result = ACTION_TAKEN_WITH_CALLBACK; } else { result = NO_ACTION_POSSIBLE; @@ -1075,7 +1137,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, uint8_t *dst_p = GRPC_SLICE_START_PTR(read_data_slice); memcpy(dst_p, stream_state->rs.read_buffer, (size_t)stream_state->rs.length_field); - free_read_buffer(s); + null_and_maybe_free_read_buffer(s); grpc_slice_buffer_init(&stream_state->rs.read_slice_buffer); grpc_slice_buffer_add(&stream_state->rs.read_slice_buffer, read_data_slice); @@ -1083,7 +1145,8 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, &stream_state->rs.read_slice_buffer, 0); *((grpc_byte_buffer **)stream_op->recv_message) = (grpc_byte_buffer *)&stream_state->rs.sbs; - grpc_closure_sched(exec_ctx, stream_op->recv_message_ready, + grpc_closure_sched(exec_ctx, + stream_op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE); stream_state->state_op_done[OP_RECV_MESSAGE] = true; oas->state.state_op_done[OP_RECV_MESSAGE] = true; @@ -1096,6 +1159,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, stream_state->rs.remaining_bytes); + stream_state->pending_read_from_cronet = true; result = ACTION_TAKEN_NO_CALLBACK; } } else if (stream_op->recv_trailing_metadata && @@ -1105,12 +1169,12 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, if (oas->s->state.rs.trailing_metadata_valid) { grpc_chttp2_incoming_metadata_buffer_publish( exec_ctx, &oas->s->state.rs.trailing_metadata, - stream_op->recv_trailing_metadata); + stream_op->payload->recv_trailing_metadata.recv_trailing_metadata); stream_state->rs.trailing_metadata_valid = false; } stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true; result = ACTION_TAKEN_NO_CALLBACK; - } else if (stream_op->cancel_error && + } else if (stream_op->cancel_stream && op_can_be_run(stream_op, s, &oas->state, OP_CANCEL_ERROR)) { CRONET_LOG(GPR_DEBUG, "running: %p OP_CANCEL_ERROR", oas); CRONET_LOG(GPR_DEBUG, "W: bidirectional_stream_cancel(%p)", s->cbs); @@ -1122,7 +1186,8 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, } stream_state->state_op_done[OP_CANCEL_ERROR] = true; if (!stream_state->cancel_error) { - stream_state->cancel_error = GRPC_ERROR_REF(stream_op->cancel_error); + stream_state->cancel_error = + GRPC_ERROR_REF(stream_op->payload->cancel_stream.cancel_error); } } else if (stream_op->on_complete && op_can_be_run(stream_op, s, &oas->state, OP_ON_COMPLETE)) { @@ -1153,15 +1218,6 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx, make a note */ if (stream_op->recv_message) stream_state->state_op_done[OP_RECV_MESSAGE_AND_ON_COMPLETE] = true; - } else if (stream_state->fail_state && !stream_state->flush_read) { - CRONET_LOG(GPR_DEBUG, "running: %p flush read", oas); - if (stream_state->rs.read_buffer && - stream_state->rs.read_buffer != stream_state->rs.grpc_header_bytes) { - gpr_free(stream_state->rs.read_buffer); - stream_state->rs.read_buffer = NULL; - } - stream_state->rs.read_buffer = gpr_malloc(4096); - stream_state->flush_read = true; } else { result = NO_ACTION_POSSIBLE; } @@ -1190,7 +1246,9 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, s->state.fail_state = s->state.flush_read = false; s->state.cancel_error = NULL; s->state.flush_cronet_when_ready = s->state.pending_write_for_trailer = false; - s->state.unprocessed_send_message = false; + s->state.pending_send_message = false; + s->state.pending_recv_trailing_metadata = false; + s->state.pending_read_from_cronet = false; s->curr_gs = gs; s->curr_ct = (grpc_cronet_transport *)gt; @@ -1209,37 +1267,33 @@ static void set_pollset_set_do_nothing(grpc_exec_ctx *exec_ctx, static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt, grpc_stream *gs, grpc_transport_stream_op *op) { CRONET_LOG(GPR_DEBUG, "perform_stream_op"); - stream_obj *s = (stream_obj *)gs; - add_to_storage(s, op); if (op->send_initial_metadata && - header_has_authority(op->send_initial_metadata->list.head)) { + header_has_authority(op->payload->send_initial_metadata + .send_initial_metadata->list.head)) { /* Cronet does not support :authority header field. We cancel the call when - this field is present in metadata */ - bidirectional_stream_header_array header_array; - bidirectional_stream_header *header; - bidirectional_stream cbs; - CRONET_LOG(GPR_DEBUG, - ":authority header is provided but not supported;" - " cancel operations"); - /* Notify application that operation is cancelled by forging trailers */ - header_array.count = 1; - header_array.capacity = 1; - header_array.headers = gpr_malloc(sizeof(bidirectional_stream_header)); - header = (bidirectional_stream_header *)header_array.headers; - header->key = "grpc-status"; - header->value = "1"; /* Return status GRPC_STATUS_CANCELLED */ - cbs.annotation = (void *)s; - s->state.state_op_done[OP_CANCEL_ERROR] = true; - on_response_trailers_received(&cbs, &header_array); - gpr_free(header_array.headers); - } else { - execute_from_storage(s); + this field is present in metadata */ + if (op->recv_initial_metadata) { + grpc_closure_sched( + exec_ctx, + op->payload->recv_initial_metadata.recv_initial_metadata_ready, + GRPC_ERROR_CANCELLED); + } + if (op->recv_message) { + grpc_closure_sched(exec_ctx, op->payload->recv_message.recv_message_ready, + GRPC_ERROR_CANCELLED); + } + grpc_closure_sched(exec_ctx, op->on_complete, GRPC_ERROR_CANCELLED); + return; } + stream_obj *s = (stream_obj *)gs; + add_to_storage(s, op); + execute_from_storage(s); } static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, grpc_stream *gs, void *and_free_memory) { stream_obj *s = (stream_obj *)gs; + null_and_maybe_free_read_buffer(s); GRPC_ERROR_UNREF(s->state.cancel_error); } diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c index 7cdbe30198..1127fff756 100644 --- a/src/core/lib/iomgr/error.c +++ b/src/core/lib/iomgr/error.c @@ -140,14 +140,16 @@ 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 @@ -182,7 +184,7 @@ static void error_destroy(grpc_error *err) { GPR_ASSERT(!grpc_error_is_special(err)); unref_errs(err); unref_strs(err); - gpr_free((void *)gpr_atm_acq_load(&err->error_string)); + gpr_free((void *)gpr_atm_acq_load(&err->atomics.error_string)); gpr_free(err); } @@ -191,15 +193,17 @@ 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); } } @@ -328,8 +332,8 @@ grpc_error *grpc_error_create(const char *file, int line, const char *desc, internal_set_time(&err, GRPC_ERROR_TIME_CREATED, gpr_now(GPR_CLOCK_REALTIME)); - gpr_atm_no_barrier_store(&err->error_string, 0); - gpr_ref_init(&err->refs, 1); + 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; } @@ -369,7 +373,7 @@ static grpc_error *copy_error_and_unref(grpc_error *in) { 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->refs)) { + } else if (gpr_ref_is_unique(&in->atomics.refs)) { out = in; } else { uint8_t new_arena_capacity = in->arena_capacity; @@ -382,10 +386,14 @@ static grpc_error *copy_error_and_unref(grpc_error *in) { #ifdef GRPC_ERROR_REFCOUNT_DEBUG gpr_log(GPR_DEBUG, "%p create copying %p", out, in); #endif - memcpy(out, in, sizeof(*in) + in->arena_size * sizeof(intptr_t)); + // 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; - gpr_atm_no_barrier_store(&out->error_string, 0); - gpr_ref_init(&out->refs, 1); ref_strs(out); ref_errs(out); GRPC_ERROR_UNREF(in); @@ -692,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; @@ -712,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_internal.h b/src/core/lib/iomgr/error_internal.h index fb4814e41f..7f204df1b2 100644 --- a/src/core/lib/iomgr/error_internal.h +++ b/src/core/lib/iomgr/error_internal.h @@ -46,14 +46,25 @@ struct grpc_linked_error { 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; + // 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; - gpr_atm error_string; + // The arena is dynamically reallocated with a grow factor of 1.5. uint8_t arena_size; uint8_t arena_capacity; intptr_t arena[0]; diff --git a/src/core/lib/iomgr/pollset_uv.c b/src/core/lib/iomgr/pollset_uv.c index af33949c69..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> @@ -61,25 +62,30 @@ gpr_mu grpc_polling_mu; 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; +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); - uv_timer_init(uv_default_loop(), &dummy_uv_handle); + dummy_uv_handle = gpr_malloc(sizeof(uv_timer_t)); + uv_timer_init(uv_default_loop(), dummy_uv_handle); grpc_pollset_work_run_loop = 1; } -static void timer_close_cb(uv_handle_t *handle) { handle->data = (void *)1; } - void grpc_pollset_global_shutdown(void) { gpr_mu_destroy(&grpc_polling_mu); - uv_close((uv_handle_t *)&dummy_uv_handle, timer_close_cb); + 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; uv_timer_init(uv_default_loop(), &pollset->timer); @@ -95,7 +101,7 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, uv_run(uv_default_loop(), UV_RUN_NOWAIT); } else { // kick the loop once - uv_timer_start(&dummy_uv_handle, dummy_timer_cb, 0, 0); + uv_timer_start(dummy_uv_handle, dummy_timer_cb, 0, 0); } grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE); } @@ -111,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) { @@ -145,7 +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); + uv_timer_start(dummy_uv_handle, dummy_timer_cb, 0, 0); return GRPC_ERROR_NONE; } 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/tcp_client_uv.c b/src/core/lib/iomgr/tcp_client_uv.c index ae66577caf..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 diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index d1bcd89af1..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) { 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/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/surface/call.c b/src/core/lib/surface/call.c index cf4410f2c0..bd34ed1b91 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -139,8 +139,8 @@ struct grpc_call { 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; @@ -155,8 +155,8 @@ struct grpc_call { bool received_initial_metadata; bool receiving_message; bool requested_final_op; - bool received_final_op; - bool sent_any_op; + gpr_atm any_ops_sent_atm; + gpr_atm received_final_op_atm; /* have we received initial metadata */ bool has_initial_md_been_received; @@ -271,7 +271,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx, GPR_TIMER_BEGIN("grpc_call_create", 0); call = gpr_zalloc(sizeof(grpc_call) + channel_stack->call_stack_size); *out_call = 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; @@ -310,7 +310,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( @@ -338,6 +338,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) { @@ -350,7 +354,7 @@ 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; @@ -432,7 +436,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); } @@ -453,7 +457,7 @@ 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); @@ -470,7 +474,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) { @@ -479,15 +483,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->sent_any_op && !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); @@ -551,56 +554,26 @@ 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; - grpc_transport_stream_op_payload payload; -} termination_closure; - -static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp, - 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, +static void done_termination(grpc_exec_ctx *exec_ctx, void *call, grpc_error *error) { - termination_closure *tc = tcp; - memset(&tc->op, 0, sizeof(tc->op)); - tc->op.payload = &tc->payload; - tc->op.cancel_stream = true; - tc->op.payload->cancel_stream.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_stream = true; + op->payload->cancel_stream.cancel_error = error; + execute_op(exec_ctx, c, op); } static grpc_error *error_from_status(grpc_status_code status, @@ -714,9 +687,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; } @@ -728,9 +699,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; } @@ -784,9 +753,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; } @@ -1033,16 +1000,13 @@ static batch_control *allocate_batch_control(grpc_call *call, const grpc_op *ops, size_t num_ops) { int slot = batch_slot_for_op(ops[0].op); - for (size_t i = 1; i < num_ops; i++) { - int op_slot = batch_slot_for_op(ops[i].op); - slot = GPR_MIN(slot, op_slot); - } batch_control *bctl = &call->active_batches[slot]; if (bctl->call != NULL) { return NULL; } memset(bctl, 0, sizeof(*bctl)); bctl->call = call; + bctl->op.payload = &call->stream_op_payload; return bctl; } @@ -1055,7 +1019,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) { @@ -1082,8 +1046,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->op.send_initial_metadata) { grpc_metadata_batch_destroy( exec_ctx, @@ -1102,20 +1064,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, @@ -1129,7 +1094,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->completion_data.notify_tag.is_closure) { /* unrefs bctl->error */ @@ -1221,7 +1185,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 +1196,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 +1257,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 +1270,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 +1295,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); } @@ -1394,11 +1351,9 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, bctl->completion_data.notify_tag.is_closure = (uint8_t)(is_notify_tag_closure != 0); - gpr_mu_lock(&call->mu); grpc_transport_stream_op *stream_op = &bctl->op; grpc_transport_stream_op_payload *stream_op_payload = &call->stream_op_payload; - memset(stream_op, 0, sizeof(*stream_op)); stream_op->covered_by_poller = true; /* rewrite batch ops into a transport op */ @@ -1513,7 +1468,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, } stream_op->send_trailing_metadata = true; call->sent_final_op = 1; - stream_op->send_trailing_metadata = + stream_op_payload->send_trailing_metadata.send_trailing_metadata = &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]; break; case GRPC_OP_SEND_STATUS_FROM_SERVER: @@ -1686,8 +1641,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; - call->sent_any_op = true; - gpr_mu_unlock(&call->mu); + gpr_atm_rel_store(&call->any_ops_sent_atm, 1); execute_op(exec_ctx, call, stream_op); @@ -1718,7 +1672,6 @@ done_with_error: if (stream_op->recv_trailing_metadata) { call->requested_final_op = 0; } - gpr_mu_unlock(&call->mu); goto done; } @@ -1767,10 +1720,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/transport/transport.c b/src/core/lib/transport/transport.c index 3b0d18c789..ce06b4d004 100644 --- a/src/core/lib/transport/transport.c +++ b/src/core/lib/transport/transport.c @@ -264,8 +264,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/cpp/common/channel_filter.h b/src/cpp/common/channel_filter.h index 79c4bab985..cf913fefe4 100644 --- a/src/cpp/common/channel_filter.h +++ b/src/cpp/common/channel_filter.h @@ -149,10 +149,22 @@ class TransportStreamOp { /// long as the TransportStreamOp object does. explicit TransportStreamOp(grpc_transport_stream_op *op) : op_(op), - send_initial_metadata_(op->send_initial_metadata), - send_trailing_metadata_(op->send_trailing_metadata), - recv_initial_metadata_(op->recv_initial_metadata), - recv_trailing_metadata_(op->recv_trailing_metadata) {} + send_initial_metadata_( + op->send_initial_metadata + ? op->payload->send_initial_metadata.send_initial_metadata + : nullptr), + send_trailing_metadata_( + op->send_trailing_metadata + ? op->payload->send_trailing_metadata.send_trailing_metadata + : nullptr), + recv_initial_metadata_( + op->recv_initial_metadata + ? op->payload->recv_initial_metadata.recv_initial_metadata + : nullptr), + recv_trailing_metadata_( + op->recv_trailing_metadata + ? op->payload->recv_trailing_metadata.recv_trailing_metadata + : nullptr) {} grpc_transport_stream_op *op() const { return op_; } @@ -160,50 +172,57 @@ class TransportStreamOp { void set_on_complete(grpc_closure *closure) { op_->on_complete = closure; } MetadataBatch *send_initial_metadata() { - return op_->send_initial_metadata == nullptr ? nullptr - : &send_initial_metadata_; + return op_->send_initial_metadata ? &send_initial_metadata_ : nullptr; } MetadataBatch *send_trailing_metadata() { - return op_->send_trailing_metadata == nullptr ? nullptr - : &send_trailing_metadata_; + return op_->send_trailing_metadata ? &send_trailing_metadata_ : nullptr; } MetadataBatch *recv_initial_metadata() { - return op_->recv_initial_metadata == nullptr ? nullptr - : &recv_initial_metadata_; + return op_->recv_initial_metadata ? &recv_initial_metadata_ : nullptr; } MetadataBatch *recv_trailing_metadata() { - return op_->recv_trailing_metadata == nullptr ? nullptr - : &recv_trailing_metadata_; + return op_->recv_trailing_metadata ? &recv_trailing_metadata_ : nullptr; } uint32_t *send_initial_metadata_flags() const { - return &op_->send_initial_metadata_flags; + return op_->send_initial_metadata + ? &op_->payload->send_initial_metadata + .send_initial_metadata_flags + : nullptr; } grpc_closure *recv_initial_metadata_ready() const { - return op_->recv_initial_metadata_ready; + return op_->recv_initial_metadata + ? op_->payload->recv_initial_metadata.recv_initial_metadata_ready + : nullptr; } void set_recv_initial_metadata_ready(grpc_closure *closure) { - op_->recv_initial_metadata_ready = closure; + op_->payload->recv_initial_metadata.recv_initial_metadata_ready = closure; } - grpc_byte_stream *send_message() const { return op_->send_message; } + grpc_byte_stream *send_message() const { + return op_->send_message ? op_->payload->send_message.send_message + : nullptr; + } void set_send_message(grpc_byte_stream *send_message) { - op_->send_message = send_message; + op_->send_message = true; + op_->payload->send_message.send_message = send_message; } /// To be called only on clients and servers, respectively. grpc_client_security_context *client_security_context() const { - return (grpc_client_security_context *)op_->context[GRPC_CONTEXT_SECURITY] + return (grpc_client_security_context *)op_->payload + ->context[GRPC_CONTEXT_SECURITY] .value; } grpc_server_security_context *server_security_context() const { - return (grpc_server_security_context *)op_->context[GRPC_CONTEXT_SECURITY] + return (grpc_server_security_context *)op_->payload + ->context[GRPC_CONTEXT_SECURITY] .value; } census_context *get_census_context() const { - return (census_context *)op_->context[GRPC_CONTEXT_TRACING].value; + return (census_context *)op_->payload->context[GRPC_CONTEXT_TRACING].value; } private: diff --git a/src/node/src/server.js b/src/node/src/server.js index 8a7eff507d..a5a0ea2642 100644 --- a/src/node/src/server.js +++ b/src/node/src/server.js @@ -728,7 +728,7 @@ var defaultHandler = { * method implementation for the provided service. */ Server.prototype.addService = function(service, implementation) { - if (!_.isObjectLike(service) || !_.isObjectLike(implementation)) { + if (!_.isObject(service) || !_.isObject(implementation)) { throw new Error('addService requires two objects as arguments'); } if (_.keys(service).length === 0) { diff --git a/src/objective-c/tests/InteropTests.m b/src/objective-c/tests/InteropTests.m index 5584246ad9..d964f53e8e 100644 --- a/src/objective-c/tests/InteropTests.m +++ b/src/objective-c/tests/InteropTests.m @@ -90,6 +90,9 @@ return nil; } +// This number indicates how many bytes of overhead does Protocol Buffers encoding add onto the +// message. The number varies as different message.proto is used on different servers. The actual +// number for each interop server is overridden in corresponding derived test classes. - (int32_t)encodingOverhead { return 0; } @@ -169,8 +172,6 @@ [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil]; } -#ifndef GRPC_COMPILE_WITH_CRONET -// TODO (mxyan): Fix this test - (void)testResponsesOverMaxSizeFailWithActionableMessage { XCTAssertNotNil(self.class.host); __weak XCTestExpectation *expectation = [self expectationWithDescription:@"ResponseOverMaxSize"]; @@ -191,7 +192,6 @@ [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil]; } -#endif - (void)testResponsesOver4MBAreAcceptedIfOptedIn { XCTAssertNotNil(self.class.host); @@ -327,8 +327,6 @@ [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil]; } -#ifndef GRPC_COMPILE_WITH_CRONET -// TODO(makdharma@): Fix this test - (void)testEmptyStreamRPC { XCTAssertNotNil(self.class.host); __weak XCTestExpectation *expectation = [self expectationWithDescription:@"EmptyStream"]; @@ -342,7 +340,6 @@ }]; [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil]; } -#endif - (void)testCancelAfterBeginRPC { XCTAssertNotNil(self.class.host); diff --git a/src/objective-c/tests/InteropTestsLocalCleartext.m b/src/objective-c/tests/InteropTestsLocalCleartext.m index b41210f50f..4987660808 100644 --- a/src/objective-c/tests/InteropTestsLocalCleartext.m +++ b/src/objective-c/tests/InteropTestsLocalCleartext.m @@ -37,6 +37,10 @@ static NSString * const kLocalCleartextHost = @"localhost:5050"; +// The Protocol Buffers encoding overhead of local interop server. Acquired +// by experiment. Adjust this when server's proto file changes. +static int32_t kLocalInteropServerOverhead = 10; + /** Tests in InteropTests.m, sending the RPCs to a local cleartext server. */ @interface InteropTestsLocalCleartext : InteropTests @end @@ -48,7 +52,7 @@ static NSString * const kLocalCleartextHost = @"localhost:5050"; } - (int32_t)encodingOverhead { - return 10; // bytes + return kLocalInteropServerOverhead; // bytes } - (void)setUp { diff --git a/src/objective-c/tests/InteropTestsLocalSSL.m b/src/objective-c/tests/InteropTestsLocalSSL.m index 1479c5896c..934d500abc 100644 --- a/src/objective-c/tests/InteropTestsLocalSSL.m +++ b/src/objective-c/tests/InteropTestsLocalSSL.m @@ -37,6 +37,10 @@ static NSString * const kLocalSSLHost = @"localhost:5051"; +// The Protocol Buffers encoding overhead of local interop server. Acquired +// by experiment. Adjust this when server's proto file changes. +static int32_t kLocalInteropServerOverhead = 10; + /** Tests in InteropTests.m, sending the RPCs to a local SSL server. */ @interface InteropTestsLocalSSL : InteropTests @end @@ -48,7 +52,7 @@ static NSString * const kLocalSSLHost = @"localhost:5051"; } - (int32_t)encodingOverhead { - return 10; // bytes + return kLocalInteropServerOverhead; // bytes } - (void)setUp { diff --git a/src/objective-c/tests/InteropTestsRemote.m b/src/objective-c/tests/InteropTestsRemote.m index 70f84753bb..9fb30aa43d 100644 --- a/src/objective-c/tests/InteropTestsRemote.m +++ b/src/objective-c/tests/InteropTestsRemote.m @@ -37,6 +37,10 @@ static NSString * const kRemoteSSLHost = @"grpc-test.sandbox.googleapis.com"; +// The Protocol Buffers encoding overhead of remote interop server. Acquired +// by experiment. Adjust this when server's proto file changes. +static int32_t kRemoteInteropServerOverhead = 12; + /** Tests in InteropTests.m, sending the RPCs to a remote SSL server. */ @interface InteropTestsRemote : InteropTests @end @@ -48,7 +52,7 @@ static NSString * const kRemoteSSLHost = @"grpc-test.sandbox.googleapis.com"; } - (int32_t)encodingOverhead { - return 12; // bytes + return kRemoteInteropServerOverhead; // bytes } @end diff --git a/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m b/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m index fab8ad8d25..005bac0a0d 100644 --- a/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m +++ b/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m @@ -37,6 +37,10 @@ static NSString * const kRemoteSSLHost = @"grpc-test.sandbox.googleapis.com"; +// The Protocol Buffers encoding overhead of remote interop server. Acquired +// by experiment. Adjust this when server's proto file changes. +static int32_t kRemoteInteropServerOverhead = 12; + /** Tests in InteropTests.m, sending the RPCs to a remote SSL server. */ @interface InteropTestsRemoteWithCronet : InteropTests @end @@ -47,4 +51,8 @@ static NSString * const kRemoteSSLHost = @"grpc-test.sandbox.googleapis.com"; return kRemoteSSLHost; } +- (int32_t)encodingOverhead { + return kRemoteInteropServerOverhead; // bytes +} + @end diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index a9f20e6d2a..52d3ffd2c4 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -33,6 +33,7 @@ CORE_SOURCE_FILES = [ 'src/core/lib/profiling/basic_timers.c', 'src/core/lib/profiling/stap_timers.c', 'src/core/lib/support/alloc.c', + 'src/core/lib/support/arena.c', 'src/core/lib/support/avl.c', 'src/core/lib/support/backoff.c', 'src/core/lib/support/cmdline.c', diff --git a/src/python/grpcio_health_checking/grpc_health/v1/health.py b/src/python/grpcio_health_checking/grpc_health/v1/health.py index f0f11cf84b..e92c2659b7 100644 --- a/src/python/grpcio_health_checking/grpc_health/v1/health.py +++ b/src/python/grpcio_health_checking/grpc_health/v1/health.py @@ -33,9 +33,10 @@ import threading import grpc from grpc_health.v1 import health_pb2 +from grpc_health.v1 import health_pb2_grpc -class HealthServicer(health_pb2.HealthServicer): +class HealthServicer(health_pb2_grpc.HealthServicer): """Servicer handling RPCs for service statuses.""" def __init__(self): diff --git a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py index 363b4c5f99..1bc8669dad 100644 --- a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py +++ b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py @@ -34,6 +34,7 @@ import grpc from grpc.framework.foundation import logging_pool from grpc_health.v1 import health from grpc_health.v1 import health_pb2 +from grpc_health.v1 import health_pb2_grpc from tests.unit.framework.common import test_constants @@ -52,11 +53,11 @@ class HealthServicerTest(unittest.TestCase): server_pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY) self._server = grpc.server(server_pool) port = self._server.add_insecure_port('[::]:0') - health_pb2.add_HealthServicer_to_server(servicer, self._server) + health_pb2_grpc.add_HealthServicer_to_server(servicer, self._server) self._server.start() channel = grpc.insecure_channel('localhost:%d' % port) - self._stub = health_pb2.HealthStub(channel) + self._stub = health_pb2_grpc.HealthStub(channel) def test_empty_service(self): request = health_pb2.HealthCheckRequest() |