diff options
Diffstat (limited to 'src')
50 files changed, 1238 insertions, 702 deletions
diff --git a/src/compiler/python_generator.cc b/src/compiler/python_generator.cc index 2649c1688d..50ee54abff 100644 --- a/src/compiler/python_generator.cc +++ b/src/compiler/python_generator.cc @@ -101,6 +101,14 @@ PrivateGenerator::PrivateGenerator(const GeneratorConfiguration& config, void PrivateGenerator::PrintAllComments(StringVector comments, grpc_generator::Printer* out) { if (comments.empty()) { + // Python requires code structures like class and def to have + // a body, even if it is just "pass" or a docstring. We need + // to ensure not to generate empty bodies. We could do something + // smarter and more sophisticated, but at the moment, if there is + // no docstring to print, we simply emit "pass" to ensure validity + // of the generated code. + out->Print("# missing associated documentation comment in .proto file\n"); + out->Print("pass\n"); return; } out->Print("\"\"\""); diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c index 16be2c70e9..ce9abdad61 100644 --- a/src/core/ext/filters/client_channel/client_channel.c +++ b/src/core/ext/filters/client_channel/client_channel.c @@ -238,14 +238,23 @@ static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx, grpc_connectivity_state state, grpc_error *error, const char *reason) { - if ((state == GRPC_CHANNEL_TRANSIENT_FAILURE || - state == GRPC_CHANNEL_SHUTDOWN) && - chand->lb_policy != NULL) { - /* cancel picks with wait_for_ready=false */ - grpc_lb_policy_cancel_picks_locked( - exec_ctx, chand->lb_policy, - /* mask= */ GRPC_INITIAL_METADATA_WAIT_FOR_READY, - /* check= */ 0, GRPC_ERROR_REF(error)); + /* TODO: Improve failure handling: + * - Make it possible for policies to return GRPC_CHANNEL_TRANSIENT_FAILURE. + * - Hand over pending picks from old policies during the switch that happens + * when resolver provides an update. */ + if (chand->lb_policy != NULL) { + if (state == GRPC_CHANNEL_TRANSIENT_FAILURE) { + /* cancel picks with wait_for_ready=false */ + grpc_lb_policy_cancel_picks_locked( + exec_ctx, chand->lb_policy, + /* mask= */ GRPC_INITIAL_METADATA_WAIT_FOR_READY, + /* check= */ 0, GRPC_ERROR_REF(error)); + } else if (state == GRPC_CHANNEL_SHUTDOWN) { + /* cancel all picks */ + grpc_lb_policy_cancel_picks_locked(exec_ctx, chand->lb_policy, + /* mask= */ 0, /* check= */ 0, + GRPC_ERROR_REF(error)); + } } grpc_connectivity_state_set(exec_ctx, &chand->state_tracker, state, error, reason); @@ -348,6 +357,33 @@ static void parse_retry_throttle_params(const grpc_json *field, void *arg) { } } +// Wrap a closure associated with \a lb_policy. The associated callback (\a +// wrapped_on_pick_closure_cb) is responsible for unref'ing \a lb_policy after +// scheduling \a wrapped_closure. +typedef struct wrapped_on_pick_closure_arg { + /* the closure instance using this struct as argument */ + grpc_closure wrapper_closure; + + /* the original closure. Usually a on_complete/notify cb for pick() and ping() + * calls against the internal RR instance, respectively. */ + grpc_closure *wrapped_closure; + + /* The policy instance related to the closure */ + grpc_lb_policy *lb_policy; +} wrapped_on_pick_closure_arg; + +// Invoke \a arg->wrapped_closure, unref \a arg->lb_policy and free \a arg. +static void wrapped_on_pick_closure_cb(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + wrapped_on_pick_closure_arg *wc_arg = arg; + GPR_ASSERT(wc_arg != NULL); + GPR_ASSERT(wc_arg->wrapped_closure != NULL); + GPR_ASSERT(wc_arg->lb_policy != NULL); + grpc_closure_run(exec_ctx, wc_arg->wrapped_closure, GRPC_ERROR_REF(error)); + GRPC_LB_POLICY_UNREF(exec_ctx, wc_arg->lb_policy, "pick_subchannel_wrapping"); + gpr_free(wc_arg); +} + static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { channel_data *chand = arg; @@ -1037,11 +1073,29 @@ static bool pick_subchannel_locked( const grpc_lb_policy_pick_args inputs = { initial_metadata, initial_metadata_flags, &calld->lb_token_mdelem, gpr_inf_future(GPR_CLOCK_MONOTONIC)}; - const bool result = grpc_lb_policy_pick_locked( - exec_ctx, lb_policy, &inputs, connected_subchannel, NULL, on_ready); + + // Wrap the user-provided callback in order to hold a strong reference to + // the LB policy for the duration of the pick. + wrapped_on_pick_closure_arg *w_on_pick_arg = + gpr_zalloc(sizeof(*w_on_pick_arg)); + grpc_closure_init(&w_on_pick_arg->wrapper_closure, + wrapped_on_pick_closure_cb, w_on_pick_arg, + grpc_schedule_on_exec_ctx); + w_on_pick_arg->wrapped_closure = on_ready; + GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel_wrapping"); + w_on_pick_arg->lb_policy = lb_policy; + const bool pick_done = grpc_lb_policy_pick_locked( + exec_ctx, lb_policy, &inputs, connected_subchannel, NULL, + &w_on_pick_arg->wrapper_closure); + if (pick_done) { + /* synchronous grpc_lb_policy_pick call. Unref the LB policy. */ + GRPC_LB_POLICY_UNREF(exec_ctx, w_on_pick_arg->lb_policy, + "pick_subchannel_wrapping"); + gpr_free(w_on_pick_arg); + } GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick_subchannel"); GPR_TIMER_END("pick_subchannel", 0); - return result; + return pick_done; } if (chand->resolver != NULL && !chand->started_resolving) { chand->started_resolving = true; diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c index e9adf98711..fb119c7fc8 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c @@ -16,6 +16,12 @@ const pb_field_t grpc_lb_v1_Duration_fields[3] = { PB_LAST_FIELD }; +const pb_field_t grpc_lb_v1_Timestamp_fields[3] = { + PB_FIELD( 1, INT64 , OPTIONAL, STATIC , FIRST, grpc_lb_v1_Timestamp, seconds, seconds, 0), + PB_FIELD( 2, INT32 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_Timestamp, nanos, seconds, 0), + PB_LAST_FIELD +}; + const pb_field_t grpc_lb_v1_LoadBalanceRequest_fields[3] = { PB_FIELD( 1, MESSAGE , OPTIONAL, STATIC , FIRST, grpc_lb_v1_LoadBalanceRequest, initial_request, initial_request, &grpc_lb_v1_InitialLoadBalanceRequest_fields), PB_FIELD( 2, MESSAGE , OPTIONAL, STATIC , OTHER, grpc_lb_v1_LoadBalanceRequest, client_stats, initial_request, &grpc_lb_v1_ClientStats_fields), @@ -27,10 +33,14 @@ const pb_field_t grpc_lb_v1_InitialLoadBalanceRequest_fields[2] = { PB_LAST_FIELD }; -const pb_field_t grpc_lb_v1_ClientStats_fields[4] = { - PB_FIELD( 1, INT64 , OPTIONAL, STATIC , FIRST, grpc_lb_v1_ClientStats, total_requests, total_requests, 0), - PB_FIELD( 2, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, client_rpc_errors, total_requests, 0), - PB_FIELD( 3, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, dropped_requests, client_rpc_errors, 0), +const pb_field_t grpc_lb_v1_ClientStats_fields[8] = { + PB_FIELD( 1, MESSAGE , OPTIONAL, STATIC , FIRST, grpc_lb_v1_ClientStats, timestamp, timestamp, &grpc_lb_v1_Timestamp_fields), + PB_FIELD( 2, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, num_calls_started, timestamp, 0), + PB_FIELD( 3, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, num_calls_finished, num_calls_started, 0), + PB_FIELD( 4, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_with_drop_for_rate_limiting, num_calls_finished, 0), + PB_FIELD( 5, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_with_drop_for_load_balancing, num_calls_finished_with_drop_for_rate_limiting, 0), + PB_FIELD( 6, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_with_client_failed_to_send, num_calls_finished_with_drop_for_load_balancing, 0), + PB_FIELD( 7, INT64 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_known_received, num_calls_finished_with_client_failed_to_send, 0), PB_LAST_FIELD }; @@ -52,11 +62,12 @@ const pb_field_t grpc_lb_v1_ServerList_fields[3] = { PB_LAST_FIELD }; -const pb_field_t grpc_lb_v1_Server_fields[5] = { +const pb_field_t grpc_lb_v1_Server_fields[6] = { PB_FIELD( 1, BYTES , OPTIONAL, STATIC , FIRST, grpc_lb_v1_Server, ip_address, ip_address, 0), PB_FIELD( 2, INT32 , OPTIONAL, STATIC , OTHER, grpc_lb_v1_Server, port, ip_address, 0), PB_FIELD( 3, STRING , OPTIONAL, STATIC , OTHER, grpc_lb_v1_Server, load_balance_token, port, 0), - PB_FIELD( 4, BOOL , OPTIONAL, STATIC , OTHER, grpc_lb_v1_Server, drop_request, load_balance_token, 0), + PB_FIELD( 4, BOOL , OPTIONAL, STATIC , OTHER, grpc_lb_v1_Server, drop_for_rate_limiting, load_balance_token, 0), + PB_FIELD( 5, BOOL , OPTIONAL, STATIC , OTHER, grpc_lb_v1_Server, drop_for_load_balancing, drop_for_rate_limiting, 0), PB_LAST_FIELD }; @@ -70,7 +81,7 @@ const pb_field_t grpc_lb_v1_Server_fields[5] = { * numbers or field sizes that are larger than what can fit in 8 or 16 bit * field descriptors. */ -PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 65536 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 65536 && pb_membersize(grpc_lb_v1_ServerList, servers) < 65536 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 65536), YOU_MUST_DEFINE_PB_FIELD_32BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server) +PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 65536 && pb_membersize(grpc_lb_v1_ClientStats, timestamp) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 65536 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 65536 && pb_membersize(grpc_lb_v1_ServerList, servers) < 65536 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 65536), YOU_MUST_DEFINE_PB_FIELD_32BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_Timestamp_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server) #endif #if !defined(PB_FIELD_16BIT) && !defined(PB_FIELD_32BIT) @@ -81,7 +92,7 @@ PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) * numbers or field sizes that are larger than what can fit in the default * 8 bit descriptors. */ -PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 256 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 256 && pb_membersize(grpc_lb_v1_ServerList, servers) < 256 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 256), YOU_MUST_DEFINE_PB_FIELD_16BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server) +PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 256 && pb_membersize(grpc_lb_v1_ClientStats, timestamp) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 256 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 256 && pb_membersize(grpc_lb_v1_ServerList, servers) < 256 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 256), YOU_MUST_DEFINE_PB_FIELD_16BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_Timestamp_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server) #endif diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h index 725aa7e386..d3ae919ec2 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h @@ -14,16 +14,6 @@ extern "C" { #endif /* Struct definitions */ -typedef struct _grpc_lb_v1_ClientStats { - bool has_total_requests; - int64_t total_requests; - bool has_client_rpc_errors; - int64_t client_rpc_errors; - bool has_dropped_requests; - int64_t dropped_requests; -/* @@protoc_insertion_point(struct:grpc_lb_v1_ClientStats) */ -} grpc_lb_v1_ClientStats; - typedef struct _grpc_lb_v1_Duration { bool has_seconds; int64_t seconds; @@ -46,11 +36,39 @@ typedef struct _grpc_lb_v1_Server { int32_t port; bool has_load_balance_token; char load_balance_token[50]; - bool has_drop_request; - bool drop_request; + bool has_drop_for_rate_limiting; + bool drop_for_rate_limiting; + bool has_drop_for_load_balancing; + bool drop_for_load_balancing; /* @@protoc_insertion_point(struct:grpc_lb_v1_Server) */ } grpc_lb_v1_Server; +typedef struct _grpc_lb_v1_Timestamp { + bool has_seconds; + int64_t seconds; + bool has_nanos; + int32_t nanos; +/* @@protoc_insertion_point(struct:grpc_lb_v1_Timestamp) */ +} grpc_lb_v1_Timestamp; + +typedef struct _grpc_lb_v1_ClientStats { + bool has_timestamp; + grpc_lb_v1_Timestamp timestamp; + bool has_num_calls_started; + int64_t num_calls_started; + bool has_num_calls_finished; + int64_t num_calls_finished; + bool has_num_calls_finished_with_drop_for_rate_limiting; + int64_t num_calls_finished_with_drop_for_rate_limiting; + bool has_num_calls_finished_with_drop_for_load_balancing; + int64_t num_calls_finished_with_drop_for_load_balancing; + bool has_num_calls_finished_with_client_failed_to_send; + int64_t num_calls_finished_with_client_failed_to_send; + bool has_num_calls_finished_known_received; + int64_t num_calls_finished_known_received; +/* @@protoc_insertion_point(struct:grpc_lb_v1_ClientStats) */ +} grpc_lb_v1_ClientStats; + typedef struct _grpc_lb_v1_InitialLoadBalanceResponse { bool has_load_balancer_delegate; char load_balancer_delegate[64]; @@ -59,6 +77,13 @@ typedef struct _grpc_lb_v1_InitialLoadBalanceResponse { /* @@protoc_insertion_point(struct:grpc_lb_v1_InitialLoadBalanceResponse) */ } grpc_lb_v1_InitialLoadBalanceResponse; +typedef struct _grpc_lb_v1_ServerList { + pb_callback_t servers; + bool has_expiration_interval; + grpc_lb_v1_Duration expiration_interval; +/* @@protoc_insertion_point(struct:grpc_lb_v1_ServerList) */ +} grpc_lb_v1_ServerList; + typedef struct _grpc_lb_v1_LoadBalanceRequest { bool has_initial_request; grpc_lb_v1_InitialLoadBalanceRequest initial_request; @@ -67,13 +92,6 @@ typedef struct _grpc_lb_v1_LoadBalanceRequest { /* @@protoc_insertion_point(struct:grpc_lb_v1_LoadBalanceRequest) */ } grpc_lb_v1_LoadBalanceRequest; -typedef struct _grpc_lb_v1_ServerList { - pb_callback_t servers; - bool has_expiration_interval; - grpc_lb_v1_Duration expiration_interval; -/* @@protoc_insertion_point(struct:grpc_lb_v1_ServerList) */ -} grpc_lb_v1_ServerList; - typedef struct _grpc_lb_v1_LoadBalanceResponse { bool has_initial_response; grpc_lb_v1_InitialLoadBalanceResponse initial_response; @@ -86,61 +104,72 @@ typedef struct _grpc_lb_v1_LoadBalanceResponse { /* Initializer values for message structs */ #define grpc_lb_v1_Duration_init_default {false, 0, false, 0} +#define grpc_lb_v1_Timestamp_init_default {false, 0, false, 0} #define grpc_lb_v1_LoadBalanceRequest_init_default {false, grpc_lb_v1_InitialLoadBalanceRequest_init_default, false, grpc_lb_v1_ClientStats_init_default} #define grpc_lb_v1_InitialLoadBalanceRequest_init_default {false, ""} -#define grpc_lb_v1_ClientStats_init_default {false, 0, false, 0, false, 0} +#define grpc_lb_v1_ClientStats_init_default {false, grpc_lb_v1_Timestamp_init_default, false, 0, false, 0, false, 0, false, 0, false, 0, false, 0} #define grpc_lb_v1_LoadBalanceResponse_init_default {false, grpc_lb_v1_InitialLoadBalanceResponse_init_default, false, grpc_lb_v1_ServerList_init_default} #define grpc_lb_v1_InitialLoadBalanceResponse_init_default {false, "", false, grpc_lb_v1_Duration_init_default} #define grpc_lb_v1_ServerList_init_default {{{NULL}, NULL}, false, grpc_lb_v1_Duration_init_default} -#define grpc_lb_v1_Server_init_default {false, {0, {0}}, false, 0, false, "", false, 0} +#define grpc_lb_v1_Server_init_default {false, {0, {0}}, false, 0, false, "", false, 0, false, 0} #define grpc_lb_v1_Duration_init_zero {false, 0, false, 0} +#define grpc_lb_v1_Timestamp_init_zero {false, 0, false, 0} #define grpc_lb_v1_LoadBalanceRequest_init_zero {false, grpc_lb_v1_InitialLoadBalanceRequest_init_zero, false, grpc_lb_v1_ClientStats_init_zero} #define grpc_lb_v1_InitialLoadBalanceRequest_init_zero {false, ""} -#define grpc_lb_v1_ClientStats_init_zero {false, 0, false, 0, false, 0} +#define grpc_lb_v1_ClientStats_init_zero {false, grpc_lb_v1_Timestamp_init_zero, false, 0, false, 0, false, 0, false, 0, false, 0, false, 0} #define grpc_lb_v1_LoadBalanceResponse_init_zero {false, grpc_lb_v1_InitialLoadBalanceResponse_init_zero, false, grpc_lb_v1_ServerList_init_zero} #define grpc_lb_v1_InitialLoadBalanceResponse_init_zero {false, "", false, grpc_lb_v1_Duration_init_zero} #define grpc_lb_v1_ServerList_init_zero {{{NULL}, NULL}, false, grpc_lb_v1_Duration_init_zero} -#define grpc_lb_v1_Server_init_zero {false, {0, {0}}, false, 0, false, "", false, 0} +#define grpc_lb_v1_Server_init_zero {false, {0, {0}}, false, 0, false, "", false, 0, false, 0} /* Field tags (for use in manual encoding/decoding) */ -#define grpc_lb_v1_ClientStats_total_requests_tag 1 -#define grpc_lb_v1_ClientStats_client_rpc_errors_tag 2 -#define grpc_lb_v1_ClientStats_dropped_requests_tag 3 #define grpc_lb_v1_Duration_seconds_tag 1 #define grpc_lb_v1_Duration_nanos_tag 2 #define grpc_lb_v1_InitialLoadBalanceRequest_name_tag 1 #define grpc_lb_v1_Server_ip_address_tag 1 #define grpc_lb_v1_Server_port_tag 2 #define grpc_lb_v1_Server_load_balance_token_tag 3 -#define grpc_lb_v1_Server_drop_request_tag 4 +#define grpc_lb_v1_Server_drop_for_rate_limiting_tag 4 +#define grpc_lb_v1_Server_drop_for_load_balancing_tag 5 +#define grpc_lb_v1_Timestamp_seconds_tag 1 +#define grpc_lb_v1_Timestamp_nanos_tag 2 +#define grpc_lb_v1_ClientStats_timestamp_tag 1 +#define grpc_lb_v1_ClientStats_num_calls_started_tag 2 +#define grpc_lb_v1_ClientStats_num_calls_finished_tag 3 +#define grpc_lb_v1_ClientStats_num_calls_finished_with_drop_for_rate_limiting_tag 4 +#define grpc_lb_v1_ClientStats_num_calls_finished_with_drop_for_load_balancing_tag 5 +#define grpc_lb_v1_ClientStats_num_calls_finished_with_client_failed_to_send_tag 6 +#define grpc_lb_v1_ClientStats_num_calls_finished_known_received_tag 7 #define grpc_lb_v1_InitialLoadBalanceResponse_load_balancer_delegate_tag 1 #define grpc_lb_v1_InitialLoadBalanceResponse_client_stats_report_interval_tag 2 -#define grpc_lb_v1_LoadBalanceRequest_initial_request_tag 1 -#define grpc_lb_v1_LoadBalanceRequest_client_stats_tag 2 #define grpc_lb_v1_ServerList_servers_tag 1 #define grpc_lb_v1_ServerList_expiration_interval_tag 3 +#define grpc_lb_v1_LoadBalanceRequest_initial_request_tag 1 +#define grpc_lb_v1_LoadBalanceRequest_client_stats_tag 2 #define grpc_lb_v1_LoadBalanceResponse_initial_response_tag 1 #define grpc_lb_v1_LoadBalanceResponse_server_list_tag 2 /* Struct field encoding specification for nanopb */ extern const pb_field_t grpc_lb_v1_Duration_fields[3]; +extern const pb_field_t grpc_lb_v1_Timestamp_fields[3]; extern const pb_field_t grpc_lb_v1_LoadBalanceRequest_fields[3]; extern const pb_field_t grpc_lb_v1_InitialLoadBalanceRequest_fields[2]; -extern const pb_field_t grpc_lb_v1_ClientStats_fields[4]; +extern const pb_field_t grpc_lb_v1_ClientStats_fields[8]; extern const pb_field_t grpc_lb_v1_LoadBalanceResponse_fields[3]; extern const pb_field_t grpc_lb_v1_InitialLoadBalanceResponse_fields[3]; extern const pb_field_t grpc_lb_v1_ServerList_fields[3]; -extern const pb_field_t grpc_lb_v1_Server_fields[5]; +extern const pb_field_t grpc_lb_v1_Server_fields[6]; /* Maximum encoded size of messages (where known) */ #define grpc_lb_v1_Duration_size 22 -#define grpc_lb_v1_LoadBalanceRequest_size 169 +#define grpc_lb_v1_Timestamp_size 22 +#define grpc_lb_v1_LoadBalanceRequest_size 226 #define grpc_lb_v1_InitialLoadBalanceRequest_size 131 -#define grpc_lb_v1_ClientStats_size 33 +#define grpc_lb_v1_ClientStats_size 90 #define grpc_lb_v1_LoadBalanceResponse_size (98 + grpc_lb_v1_ServerList_size) #define grpc_lb_v1_InitialLoadBalanceResponse_size 90 /* grpc_lb_v1_ServerList_size depends on runtime parameters */ -#define grpc_lb_v1_Server_size 83 +#define grpc_lb_v1_Server_size 85 /* Message IDs (where set with "msgid" option) */ #ifdef PB_MSGID diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.c b/src/core/ext/filters/client_channel/lb_policy_factory.c index e2af216b89..89b8bf8951 100644 --- a/src/core/ext/filters/client_channel/lb_policy_factory.c +++ b/src/core/ext/filters/client_channel/lb_policy_factory.c @@ -36,16 +36,18 @@ #include <grpc/support/alloc.h> #include <grpc/support/string_util.h> +#include "src/core/lib/channel/channel_args.h" + #include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include "src/core/ext/filters/client_channel/parse_address.h" grpc_lb_addresses* grpc_lb_addresses_create( size_t num_addresses, const grpc_lb_user_data_vtable* user_data_vtable) { - grpc_lb_addresses* addresses = gpr_malloc(sizeof(grpc_lb_addresses)); + grpc_lb_addresses* addresses = gpr_zalloc(sizeof(grpc_lb_addresses)); addresses->num_addresses = num_addresses; addresses->user_data_vtable = user_data_vtable; const size_t addresses_size = sizeof(grpc_lb_address) * num_addresses; - addresses->addresses = gpr_malloc(addresses_size); - memset(addresses->addresses, 0, addresses_size); + addresses->addresses = gpr_zalloc(addresses_size); return addresses; } @@ -69,7 +71,7 @@ grpc_lb_addresses* grpc_lb_addresses_copy(const grpc_lb_addresses* addresses) { void grpc_lb_addresses_set_address(grpc_lb_addresses* addresses, size_t index, void* address, size_t address_len, - bool is_balancer, char* balancer_name, + bool is_balancer, const char* balancer_name, void* user_data) { GPR_ASSERT(index < addresses->num_addresses); if (user_data != NULL) GPR_ASSERT(addresses->user_data_vtable != NULL); @@ -77,10 +79,22 @@ void grpc_lb_addresses_set_address(grpc_lb_addresses* addresses, size_t index, memcpy(target->address.addr, address, address_len); target->address.len = address_len; target->is_balancer = is_balancer; - target->balancer_name = balancer_name; + target->balancer_name = gpr_strdup(balancer_name); target->user_data = user_data; } +bool grpc_lb_addresses_set_address_from_uri(grpc_lb_addresses* addresses, + size_t index, const grpc_uri* uri, + bool is_balancer, + const char* balancer_name, + void* user_data) { + grpc_resolved_address address; + if (!grpc_parse_uri(uri, &address)) return false; + grpc_lb_addresses_set_address(addresses, index, address.addr, address.len, + is_balancer, balancer_name, user_data); + return true; +} + int grpc_lb_addresses_cmp(const grpc_lb_addresses* addresses1, const grpc_lb_addresses* addresses2) { if (addresses1->num_addresses > addresses2->num_addresses) return 1; @@ -147,6 +161,15 @@ grpc_arg grpc_lb_addresses_create_channel_arg( return arg; } +grpc_lb_addresses* grpc_lb_addresses_find_channel_arg( + const grpc_channel_args* channel_args) { + const grpc_arg* lb_addresses_arg = + grpc_channel_args_find(channel_args, GRPC_ARG_LB_ADDRESSES); + if (lb_addresses_arg == NULL || lb_addresses_arg->type != GRPC_ARG_POINTER) + return NULL; + return lb_addresses_arg->value.pointer.p; +} + void grpc_lb_policy_factory_ref(grpc_lb_policy_factory* factory) { factory->vtable->ref(factory); } diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.h b/src/core/ext/filters/client_channel/lb_policy_factory.h index 81ab12ec8f..9d6c0fc139 100644 --- a/src/core/ext/filters/client_channel/lb_policy_factory.h +++ b/src/core/ext/filters/client_channel/lb_policy_factory.h @@ -34,12 +34,13 @@ #ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_FACTORY_H #define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_FACTORY_H -#include "src/core/ext/filters/client_channel/client_channel_factory.h" -#include "src/core/ext/filters/client_channel/lb_policy.h" - #include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/resolve_address.h" +#include "src/core/ext/filters/client_channel/client_channel_factory.h" +#include "src/core/ext/filters/client_channel/lb_policy.h" +#include "src/core/ext/filters/client_channel/uri_parser.h" + // Channel arg key for grpc_lb_addresses. #define GRPC_ARG_LB_ADDRESSES "grpc.lb_addresses" @@ -88,9 +89,18 @@ grpc_lb_addresses *grpc_lb_addresses_copy(const grpc_lb_addresses *addresses); * Takes ownership of \a balancer_name. */ void grpc_lb_addresses_set_address(grpc_lb_addresses *addresses, size_t index, void *address, size_t address_len, - bool is_balancer, char *balancer_name, + bool is_balancer, const char *balancer_name, void *user_data); +/** Sets the value of the address at index \a index of \a addresses from \a uri. + * Returns true upon success, false otherwise. Takes ownership of \a + * balancer_name. */ +bool grpc_lb_addresses_set_address_from_uri(grpc_lb_addresses *addresses, + size_t index, const grpc_uri *uri, + bool is_balancer, + const char *balancer_name, + void *user_data); + /** Compares \a addresses1 and \a addresses2. */ int grpc_lb_addresses_cmp(const grpc_lb_addresses *addresses1, const grpc_lb_addresses *addresses2); @@ -103,6 +113,10 @@ void grpc_lb_addresses_destroy(grpc_exec_ctx *exec_ctx, grpc_arg grpc_lb_addresses_create_channel_arg( const grpc_lb_addresses *addresses); +/** Returns the \a grpc_lb_addresses instance in \a channel_args or NULL */ +grpc_lb_addresses *grpc_lb_addresses_find_channel_arg( + const grpc_channel_args *channel_args); + /** Arguments passed to LB policies. */ typedef struct grpc_lb_policy_args { grpc_client_channel_factory *client_channel_factory; diff --git a/src/core/ext/filters/client_channel/parse_address.c b/src/core/ext/filters/client_channel/parse_address.c index 0c97062075..edc6ce697d 100644 --- a/src/core/ext/filters/client_channel/parse_address.c +++ b/src/core/ext/filters/client_channel/parse_address.c @@ -48,7 +48,12 @@ #ifdef GRPC_HAVE_UNIX_SOCKET -int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr) { +bool grpc_parse_unix(const grpc_uri *uri, + grpc_resolved_address *resolved_addr) { + if (strcmp("unix", uri->scheme) != 0) { + gpr_log(GPR_ERROR, "Expected 'unix' scheme, got '%s'", uri->scheme); + return false; + } struct sockaddr_un *un = (struct sockaddr_un *)resolved_addr->addr; const size_t maxlen = sizeof(un->sun_path); const size_t path_len = strnlen(uri->path, maxlen); @@ -61,21 +66,29 @@ int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr) { #else /* GRPC_HAVE_UNIX_SOCKET */ -int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr) { abort(); } +bool grpc_parse_unix(const grpc_uri *uri, + grpc_resolved_address *resolved_addr) { + abort(); +} #endif /* GRPC_HAVE_UNIX_SOCKET */ -int parse_ipv4(grpc_uri *uri, grpc_resolved_address *resolved_addr) { +bool grpc_parse_ipv4(const grpc_uri *uri, + grpc_resolved_address *resolved_addr) { + if (strcmp("ipv4", uri->scheme) != 0) { + gpr_log(GPR_ERROR, "Expected 'ipv4' scheme, got '%s'", uri->scheme); + return false; + } const char *host_port = uri->path; char *host; char *port; int port_num; - int result = 0; + bool result = false; struct sockaddr_in *in = (struct sockaddr_in *)resolved_addr->addr; if (*host_port == '/') ++host_port; if (!gpr_split_host_port(host_port, &host, &port)) { - return 0; + return false; } memset(resolved_addr, 0, sizeof(grpc_resolved_address)); @@ -98,14 +111,19 @@ int parse_ipv4(grpc_uri *uri, grpc_resolved_address *resolved_addr) { goto done; } - result = 1; + result = true; done: gpr_free(host); gpr_free(port); return result; } -int parse_ipv6(grpc_uri *uri, grpc_resolved_address *resolved_addr) { +bool grpc_parse_ipv6(const grpc_uri *uri, + grpc_resolved_address *resolved_addr) { + if (strcmp("ipv6", uri->scheme) != 0) { + gpr_log(GPR_ERROR, "Expected 'ipv6' scheme, got '%s'", uri->scheme); + return false; + } const char *host_port = uri->path; char *host; char *port; @@ -168,3 +186,15 @@ done: gpr_free(port); return result; } + +bool grpc_parse_uri(const grpc_uri *uri, grpc_resolved_address *resolved_addr) { + if (strcmp("unix", uri->scheme) == 0) { + return grpc_parse_unix(uri, resolved_addr); + } else if (strcmp("ipv4", uri->scheme) == 0) { + return grpc_parse_ipv4(uri, resolved_addr); + } else if (strcmp("ipv6", uri->scheme) == 0) { + return grpc_parse_ipv6(uri, resolved_addr); + } + gpr_log(GPR_ERROR, "Can't parse scheme '%s'", uri->scheme); + return false; +} diff --git a/src/core/ext/filters/client_channel/parse_address.h b/src/core/ext/filters/client_channel/parse_address.h index c8d77baa00..fa7ea33a00 100644 --- a/src/core/ext/filters/client_channel/parse_address.h +++ b/src/core/ext/filters/client_channel/parse_address.h @@ -39,16 +39,19 @@ #include "src/core/ext/filters/client_channel/uri_parser.h" #include "src/core/lib/iomgr/resolve_address.h" -/** Populate \a addr and \a len from \a uri, whose path is expected to contain a +/** Populate \a resolved_addr from \a uri, whose path is expected to contain a * unix socket path. Returns true upon success. */ -int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr); +bool grpc_parse_unix(const grpc_uri *uri, grpc_resolved_address *resolved_addr); -/** Populate /a addr and \a len from \a uri, whose path is expected to contain a - * host:port pair. Returns true upon success. */ -int parse_ipv4(grpc_uri *uri, grpc_resolved_address *resolved_addr); +/** Populate \a resolved_addr from \a uri, whose path is expected to contain an + * IPv4 host:port pair. Returns true upon success. */ +bool grpc_parse_ipv4(const grpc_uri *uri, grpc_resolved_address *resolved_addr); -/** Populate /a addr and \a len from \a uri, whose path is expected to contain a - * host:port pair. Returns true upon success. */ -int parse_ipv6(grpc_uri *uri, grpc_resolved_address *resolved_addr); +/** Populate \a resolved_addr from \a uri, whose path is expected to contain an + * IPv6 host:port pair. Returns true upon success. */ +bool grpc_parse_ipv6(const grpc_uri *uri, grpc_resolved_address *resolved_addr); + +/** Populate \a resolved_addr from \a uri. Returns true upon success. */ +bool grpc_parse_uri(const grpc_uri *uri, grpc_resolved_address *resolved_addr); #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_PARSE_ADDRESS_H */ diff --git a/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c b/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c index 54f020d691..4d7d878c23 100644 --- a/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c +++ b/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c @@ -157,8 +157,8 @@ static void do_nothing(void *ignored) {} static grpc_resolver *sockaddr_create(grpc_exec_ctx *exec_ctx, grpc_resolver_args *args, - int parse(grpc_uri *uri, - grpc_resolved_address *dst)) { + bool parse(const grpc_uri *uri, + grpc_resolved_address *dst)) { if (0 != strcmp(args->uri->authority, "")) { gpr_log(GPR_ERROR, "authority based uri's not supported by the %s scheme", args->uri->scheme); @@ -209,7 +209,7 @@ static void sockaddr_factory_unref(grpc_resolver_factory *factory) {} static grpc_resolver *name##_factory_create_resolver( \ grpc_exec_ctx *exec_ctx, grpc_resolver_factory *factory, \ grpc_resolver_args *args) { \ - return sockaddr_create(exec_ctx, args, parse_##name); \ + return sockaddr_create(exec_ctx, args, grpc_parse_##name); \ } \ static const grpc_resolver_factory_vtable name##_factory_vtable = { \ sockaddr_factory_ref, sockaddr_factory_unref, \ diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c index 9a7a7a0ee5..967e571221 100644 --- a/src/core/ext/filters/client_channel/subchannel.c +++ b/src/core/ext/filters/client_channel/subchannel.c @@ -59,9 +59,9 @@ #define INTERNAL_REF_BITS 16 #define STRONG_REF_MASK (~(gpr_atm)((1 << INTERNAL_REF_BITS) - 1)) -#define GRPC_SUBCHANNEL_MIN_CONNECT_TIMEOUT_SECONDS 20 #define GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS 1 #define GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER 1.6 +#define GRPC_SUBCHANNEL_RECONNECT_MIN_BACKOFF_SECONDS 20 #define GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS 120 #define GRPC_SUBCHANNEL_RECONNECT_JITTER 0.2 @@ -353,8 +353,8 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx, "subchannel"); int initial_backoff_ms = GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS * 1000; + int min_backoff_ms = GRPC_SUBCHANNEL_RECONNECT_MIN_BACKOFF_SECONDS * 1000; int max_backoff_ms = GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS * 1000; - int min_backoff_ms = GRPC_SUBCHANNEL_MIN_CONNECT_TIMEOUT_SECONDS * 1000; bool fixed_reconnect_backoff = false; if (c->args) { for (size_t i = 0; i < c->args->num_args; i++) { @@ -366,6 +366,12 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx, &c->args->args[i], (grpc_integer_options){initial_backoff_ms, 100, INT_MAX}); } else if (0 == strcmp(c->args->args[i].key, + GRPC_ARG_MIN_RECONNECT_BACKOFF_MS)) { + fixed_reconnect_backoff = false; + min_backoff_ms = grpc_channel_arg_get_integer( + &c->args->args[i], + (grpc_integer_options){min_backoff_ms, 100, INT_MAX}); + } else if (0 == strcmp(c->args->args[i].key, GRPC_ARG_MAX_RECONNECT_BACKOFF_MS)) { fixed_reconnect_backoff = false; max_backoff_ms = grpc_channel_arg_get_integer( @@ -797,13 +803,7 @@ static void grpc_uri_to_sockaddr(grpc_exec_ctx *exec_ctx, const char *uri_str, grpc_resolved_address *addr) { grpc_uri *uri = grpc_uri_parse(exec_ctx, uri_str, 0 /* suppress_errors */); GPR_ASSERT(uri != NULL); - if (strcmp(uri->scheme, "ipv4") == 0) { - GPR_ASSERT(parse_ipv4(uri, addr)); - } else if (strcmp(uri->scheme, "ipv6") == 0) { - GPR_ASSERT(parse_ipv6(uri, addr)); - } else { - GPR_ASSERT(parse_unix(uri, addr)); - } + if (!grpc_parse_uri(uri, addr)) memset(addr, 0, sizeof(*addr)); grpc_uri_destroy(uri); } diff --git a/src/core/ext/filters/client_channel/uri_parser.c b/src/core/ext/filters/client_channel/uri_parser.c index 01b99911aa..b233d835cb 100644 --- a/src/core/ext/filters/client_channel/uri_parser.c +++ b/src/core/ext/filters/client_channel/uri_parser.c @@ -50,7 +50,7 @@ #define NOT_SET (~(size_t)0) static grpc_uri *bad_uri(const char *uri_text, size_t pos, const char *section, - int suppress_errors) { + bool suppress_errors) { char *line_prefix; size_t pfx_len; @@ -83,6 +83,11 @@ static char *decode_and_copy_component(grpc_exec_ctx *exec_ctx, const char *src, return out; } +static bool valid_hex(char c) { + return ((c >= 'a') && (c <= 'f')) || ((c >= 'A') && (c <= 'F')) || + ((c >= '0') && (c <= '9')); +} + /** Returns how many chars to advance if \a uri_text[i] begins a valid \a pchar * production. If \a uri_text[i] introduces an invalid \a pchar (such as percent * sign not followed by two hex digits), NOT_SET is returned. */ @@ -93,27 +98,36 @@ static size_t parse_pchar(const char *uri_text, size_t i) { * sub-delims = "!" / "$" / "&" / "'" / "(" / ")" / "*" / "+" / "," / ";" / "=" */ char c = uri_text[i]; - if (((c >= 'A') && (c <= 'Z')) || ((c >= 'a') && (c <= 'z')) || - ((c >= '0') && (c <= '9')) || - (c == '-' || c == '.' || c == '_' || c == '~') || /* unreserved */ - (c == '!' || c == '$' || c == '&' || c == '\'' || c == '$' || c == '&' || - c == '(' || c == ')' || c == '*' || c == '+' || c == ',' || c == ';' || - c == '=') /* sub-delims */) { - return 1; - } - if (c == '%') { /* pct-encoded */ - size_t j; - if (uri_text[i + 1] == 0 || uri_text[i + 2] == 0) { - return NOT_SET; - } - for (j = i + 1; j < 2; j++) { - c = uri_text[j]; - if (!(((c >= '0') && (c <= '9')) || ((c >= 'a') && (c <= 'f')) || - ((c >= 'A') && (c <= 'F')))) { - return NOT_SET; + switch (c) { + default: + if (((c >= 'a') && (c <= 'z')) || ((c >= 'A') && (c <= 'Z')) || + ((c >= '0') && (c <= '9'))) { + return 1; } - } - return 2; + break; + case ':': + case '@': + case '-': + case '.': + case '_': + case '~': + case '!': + case '$': + case '&': + case '\'': + case '(': + case ')': + case '*': + case '+': + case ',': + case ';': + case '=': + return 1; + case '%': /* pct-encoded */ + if (valid_hex(uri_text[i + 1]) && valid_hex(uri_text[i + 2])) { + return 2; + } + return NOT_SET; } return 0; } @@ -183,7 +197,7 @@ static void parse_query_parts(grpc_uri *uri) { } grpc_uri *grpc_uri_parse(grpc_exec_ctx *exec_ctx, const char *uri_text, - int suppress_errors) { + bool suppress_errors) { grpc_uri *uri; size_t scheme_begin = 0; size_t scheme_end = NOT_SET; diff --git a/src/core/ext/filters/client_channel/uri_parser.h b/src/core/ext/filters/client_channel/uri_parser.h index 2698d448d8..b889040b16 100644 --- a/src/core/ext/filters/client_channel/uri_parser.h +++ b/src/core/ext/filters/client_channel/uri_parser.h @@ -53,7 +53,7 @@ typedef struct { /** parse a uri, return NULL on failure */ grpc_uri *grpc_uri_parse(grpc_exec_ctx *exec_ctx, const char *uri_text, - int suppress_errors); + bool suppress_errors); /** return the part of a query string after the '=' in "?key=xxx&...", or NULL * if key is not present */ diff --git a/src/core/ext/filters/http/client/http_client_filter.c b/src/core/ext/filters/http/client/http_client_filter.c index b896c7e7bb..bf5fbc7da7 100644 --- a/src/core/ext/filters/http/client/http_client_filter.c +++ b/src/core/ext/filters/http/client/http_client_filter.c @@ -220,8 +220,11 @@ static void continue_send_message(grpc_exec_ctx *exec_ctx, call_data *calld = elem->call_data; uint8_t *wrptr = calld->payload_bytes; while (grpc_byte_stream_next( - exec_ctx, calld->send_op->payload->send_message.send_message, - &calld->incoming_slice, ~(size_t)0, &calld->got_slice)) { + exec_ctx, calld->send_op->payload->send_message.send_message, ~(size_t)0, + &calld->got_slice)) { + grpc_byte_stream_pull(exec_ctx, + calld->send_op->payload->send_message.send_message, + &calld->incoming_slice); if (GRPC_SLICE_LENGTH(calld->incoming_slice) > 0) { memcpy(wrptr, GRPC_SLICE_START_PTR(calld->incoming_slice), GRPC_SLICE_LENGTH(calld->incoming_slice)); @@ -239,6 +242,13 @@ static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { grpc_call_element *elem = elemp; call_data *calld = elem->call_data; calld->send_message_blocked = false; + if (GRPC_ERROR_NONE != + grpc_byte_stream_pull(exec_ctx, + calld->send_op->payload->send_message.send_message, + &calld->incoming_slice)) { + /* Should never reach here */ + abort(); + } grpc_slice_buffer_add(&calld->slices, calld->incoming_slice); if (calld->send_length == calld->slices.length) { /* Pass down the original send_message op that was blocked.*/ @@ -313,7 +323,6 @@ static grpc_error *hc_mutate_op(grpc_exec_ctx *exec_ctx, estimated_len += grpc_base64_estimate_encoded_size( op->payload->send_message.send_message->length, k_url_safe, k_multi_line); - estimated_len += 1; /* for the trailing 0 */ grpc_slice path_with_query_slice = grpc_slice_malloc(estimated_len); /* memcopy individual pieces into this slice */ @@ -335,7 +344,7 @@ static grpc_error *hc_mutate_op(grpc_exec_ctx *exec_ctx, char *t = (char *)GRPC_SLICE_START_PTR(path_with_query_slice); /* safe to use strlen since base64_encode will always add '\0' */ path_with_query_slice = - grpc_slice_sub(path_with_query_slice, 0, strlen(t)); + grpc_slice_sub_no_ref(path_with_query_slice, 0, strlen(t)); /* substitute previous path with the new path+query */ grpc_mdelem mdelem_path_and_query = grpc_mdelem_from_slices( @@ -349,7 +358,6 @@ static grpc_error *hc_mutate_op(grpc_exec_ctx *exec_ctx, calld->on_complete = op->on_complete; op->on_complete = &calld->hc_on_complete; op->send_message = false; - grpc_slice_unref_internal(exec_ctx, path_with_query_slice); } else { /* Not all data is available. Fall back to POST. */ gpr_log(GPR_DEBUG, diff --git a/src/core/ext/filters/http/message_compress/message_compress_filter.c b/src/core/ext/filters/http/message_compress/message_compress_filter.c index 2b2f3549bd..f414a60eee 100644 --- a/src/core/ext/filters/http/message_compress/message_compress_filter.c +++ b/src/core/ext/filters/http/message_compress/message_compress_filter.c @@ -239,6 +239,13 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx, static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { grpc_call_element *elem = elemp; call_data *calld = elem->call_data; + if (GRPC_ERROR_NONE != + grpc_byte_stream_pull(exec_ctx, + calld->send_op->payload->send_message.send_message, + &calld->incoming_slice)) { + /* Should never reach here */ + abort(); + } grpc_slice_buffer_add(&calld->slices, calld->incoming_slice); if (calld->send_length == calld->slices.length) { finish_send_message(exec_ctx, elem); @@ -251,8 +258,11 @@ static void continue_send_message(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { call_data *calld = elem->call_data; while (grpc_byte_stream_next( - exec_ctx, calld->send_op->payload->send_message.send_message, - &calld->incoming_slice, ~(size_t)0, &calld->got_slice)) { + exec_ctx, calld->send_op->payload->send_message.send_message, ~(size_t)0, + &calld->got_slice)) { + grpc_byte_stream_pull(exec_ctx, + calld->send_op->payload->send_message.send_message, + &calld->incoming_slice); grpc_slice_buffer_add(&calld->slices, calld->incoming_slice); if (calld->send_length == calld->slices.length) { finish_send_message(exec_ctx, elem); diff --git a/src/core/ext/filters/load_reporting/load_reporting.c b/src/core/ext/filters/load_reporting/load_reporting.c index 3b5d2103c1..4e9d0937ae 100644 --- a/src/core/ext/filters/load_reporting/load_reporting.c +++ b/src/core/ext/filters/load_reporting/load_reporting.c @@ -47,23 +47,6 @@ #include "src/core/lib/surface/call.h" #include "src/core/lib/surface/channel_init.h" -static void destroy_lr_cost_context(void *c) { - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_load_reporting_cost_context *cost_ctx = c; - for (size_t i = 0; i < cost_ctx->values_count; ++i) { - grpc_slice_unref_internal(&exec_ctx, cost_ctx->values[i]); - } - grpc_exec_ctx_finish(&exec_ctx); - gpr_free(cost_ctx->values); - gpr_free(cost_ctx); -} - -void grpc_call_set_load_reporting_cost_context( - grpc_call *call, grpc_load_reporting_cost_context *ctx) { - grpc_call_context_set(call, GRPC_CONTEXT_LR_COST, ctx, - destroy_lr_cost_context); -} - static bool is_load_reporting_enabled(const grpc_channel_args *a) { return grpc_channel_arg_get_bool( grpc_channel_args_find(a, GRPC_ARG_ENABLE_LOAD_REPORTING), false); diff --git a/src/core/ext/filters/load_reporting/load_reporting_filter.c b/src/core/ext/filters/load_reporting/load_reporting_filter.c index 57b25d0651..75a9a56687 100644 --- a/src/core/ext/filters/load_reporting/load_reporting_filter.c +++ b/src/core/ext/filters/load_reporting/load_reporting_filter.c @@ -48,6 +48,8 @@ typedef struct call_data { intptr_t id; /**< an id unique to the call */ + bool have_trailing_md_string; + grpc_slice trailing_md_string; bool have_initial_md_string; grpc_slice initial_md_string; bool have_service_method; @@ -140,6 +142,9 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, if (calld->have_initial_md_string) { grpc_slice_unref_internal(exec_ctx, calld->initial_md_string); } + if (calld->have_trailing_md_string) { + grpc_slice_unref_internal(exec_ctx, calld->trailing_md_string); + } if (calld->have_service_method) { grpc_slice_unref_internal(exec_ctx, calld->service_method); } @@ -183,6 +188,18 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, */ } +static grpc_filtered_mdelem lr_trailing_md_filter(grpc_exec_ctx *exec_ctx, + void *user_data, + grpc_mdelem md) { + grpc_call_element *elem = user_data; + call_data *calld = elem->call_data; + if (grpc_slice_eq(GRPC_MDKEY(md), GRPC_MDSTR_LB_COST_BIN)) { + calld->trailing_md_string = GRPC_MDVALUE(md); + return GRPC_FILTERED_REMOVE(); + } + return GRPC_FILTERED_MDELEM(md); +} + static void lr_start_transport_stream_op_batch( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *op) { @@ -190,13 +207,21 @@ static void lr_start_transport_stream_op_batch( call_data *calld = elem->call_data; if (op->recv_initial_metadata) { + /* substitute our callback for the higher callback */ calld->recv_initial_metadata = op->payload->recv_initial_metadata.recv_initial_metadata; - /* substitute our callback for the higher callback */ calld->ops_recv_initial_metadata_ready = op->payload->recv_initial_metadata.recv_initial_metadata_ready; op->payload->recv_initial_metadata.recv_initial_metadata_ready = &calld->on_initial_md_ready; + } else if (op->send_trailing_metadata) { + GRPC_LOG_IF_ERROR( + "grpc_metadata_batch_filter", + grpc_metadata_batch_filter( + exec_ctx, + op->payload->send_trailing_metadata.send_trailing_metadata, + lr_trailing_md_filter, elem, + "LR trailing metadata filtering error")); } grpc_call_next_op(exec_ctx, elem, op); diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 7979883418..5c5175f566 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -44,6 +44,7 @@ #include <grpc/support/string_util.h> #include <grpc/support/useful.h> +#include "src/core/ext/transport/chttp2/transport/frame_data.h" #include "src/core/ext/transport/chttp2/transport/internal.h" #include "src/core/ext/transport/chttp2/transport/varint.h" #include "src/core/lib/channel/channel_args.h" @@ -129,6 +130,11 @@ static void incoming_byte_stream_update_flow_control(grpc_exec_ctx *exec_ctx, static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx, void *byte_stream, grpc_error *error_ignored); +static void incoming_byte_stream_publish_error( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs, + grpc_error *error); +static void incoming_byte_stream_unref(grpc_exec_ctx *exec_ctx, + grpc_chttp2_incoming_byte_stream *bs); static void benign_reclaimer_locked(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); @@ -174,6 +180,9 @@ static void finish_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, static void keepalive_watchdog_fired_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error); +static void reset_byte_stream(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error); + /******************************************************************************* * CONSTRUCTION/DESTRUCTION/REFCOUNTING */ @@ -656,7 +665,6 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, /* We reserve one 'active stream' that's dropped when the stream is read-closed. The others are for incoming_byte_streams that are actively reading */ - gpr_ref_init(&s->active_streams, 1); GRPC_CHTTP2_STREAM_REF(s, "chttp2"); grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[0], arena); @@ -666,6 +674,11 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, s->deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); grpc_closure_init(&s->complete_fetch_locked, complete_fetch_locked, s, grpc_schedule_on_exec_ctx); + grpc_slice_buffer_init(&s->unprocessed_incoming_frames_buffer); + grpc_slice_buffer_init(&s->frame_storage); + s->pending_byte_stream = false; + grpc_closure_init(&s->reset_byte_stream, reset_byte_stream, s, + grpc_combiner_scheduler(t->combiner, false)); GRPC_CHTTP2_REF_TRANSPORT(t, "stream"); @@ -683,7 +696,6 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp, grpc_error *error) { - grpc_byte_stream *bs; grpc_chttp2_stream *s = sp; grpc_chttp2_transport *t = s->t; @@ -694,9 +706,9 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp, GPR_ASSERT(grpc_chttp2_stream_map_find(&t->stream_map, s->id) == NULL); } - while ((bs = grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames))) { - incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE); - } + grpc_slice_buffer_destroy_internal(exec_ctx, + &s->unprocessed_incoming_frames_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &s->frame_storage); grpc_chttp2_list_remove_stalled_by_transport(t, s); grpc_chttp2_list_remove_stalled_by_stream(t, s); @@ -723,6 +735,7 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp, grpc_slice_buffer_destroy_internal(exec_ctx, &s->flow_controlled_buffer); GRPC_ERROR_UNREF(s->read_closed_error); GRPC_ERROR_UNREF(s->write_closed_error); + GRPC_ERROR_UNREF(s->byte_stream_error); if (s->incoming_window_delta > 0) { GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA( @@ -1176,8 +1189,9 @@ static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx, s->fetching_send_message = NULL; return; /* early out */ } else if (grpc_byte_stream_next(exec_ctx, s->fetching_send_message, - &s->fetching_slice, UINT32_MAX, - &s->complete_fetch_locked)) { + UINT32_MAX, &s->complete_fetch_locked)) { + grpc_byte_stream_pull(exec_ctx, s->fetching_send_message, + &s->fetching_slice); add_fetched_slice_locked(exec_ctx, t, s); } } @@ -1188,9 +1202,15 @@ static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs, grpc_chttp2_stream *s = gs; grpc_chttp2_transport *t = s->t; if (error == GRPC_ERROR_NONE) { - add_fetched_slice_locked(exec_ctx, t, s); - continue_fetching_send_locked(exec_ctx, t, s); - } else { + error = grpc_byte_stream_pull(exec_ctx, s->fetching_send_message, + &s->fetching_slice); + if (error == GRPC_ERROR_NONE) { + add_fetched_slice_locked(exec_ctx, t, s); + continue_fetching_send_locked(exec_ctx, t, s); + } + } + + if (error != GRPC_ERROR_NONE) { /* TODO(ctiller): what to do here */ abort(); } @@ -1422,12 +1442,20 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, } if (op->recv_message) { + size_t already_received; GPR_ASSERT(s->recv_message_ready == NULL); + GPR_ASSERT(!s->pending_byte_stream); s->recv_message_ready = op_payload->recv_message.recv_message_ready; s->recv_message = op_payload->recv_message.recv_message; - if (s->id != 0 && - (s->incoming_frames.head == NULL || s->incoming_frames.head->is_tail)) { - incoming_byte_stream_update_flow_control(exec_ctx, t, s, 5, 0); + if (s->id != 0) { + if (s->pending_byte_stream) { + already_received = s->frame_storage.length; + } else { + already_received = s->frame_storage.length + + s->unprocessed_incoming_frames_buffer.length; + } + incoming_byte_stream_update_flow_control(exec_ctx, t, s, 5, + already_received); } grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s); } @@ -1615,13 +1643,13 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt, void grpc_chttp2_maybe_complete_recv_initial_metadata(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s) { - grpc_byte_stream *bs; if (s->recv_initial_metadata_ready != NULL && s->published_metadata[0] != GRPC_METADATA_NOT_PUBLISHED) { if (s->seen_error) { - while ((bs = grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames)) != - NULL) { - incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &s->frame_storage); + if (!s->pending_byte_stream) { + grpc_slice_buffer_reset_and_unref_internal( + exec_ctx, &s->unprocessed_incoming_frames_buffer); } } grpc_chttp2_incoming_metadata_buffer_publish( @@ -1634,39 +1662,65 @@ void grpc_chttp2_maybe_complete_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void grpc_chttp2_maybe_complete_recv_message(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s) { - grpc_byte_stream *bs; + grpc_error *error = GRPC_ERROR_NONE; if (s->recv_message_ready != NULL) { - while (s->final_metadata_requested && s->seen_error && - (bs = grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames)) != - NULL) { - incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE); + *s->recv_message = NULL; + if (s->final_metadata_requested && s->seen_error) { + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &s->frame_storage); + if (!s->pending_byte_stream) { + grpc_slice_buffer_reset_and_unref_internal( + exec_ctx, &s->unprocessed_incoming_frames_buffer); + } } - if (s->incoming_frames.head != NULL) { - *s->recv_message = - grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames); - GPR_ASSERT(*s->recv_message != NULL); + if (!s->pending_byte_stream) { + while (s->unprocessed_incoming_frames_buffer.length > 0 || + s->frame_storage.length > 0) { + if (s->unprocessed_incoming_frames_buffer.length == 0) { + grpc_slice_buffer_swap(&s->unprocessed_incoming_frames_buffer, + &s->frame_storage); + } + error = grpc_deframe_unprocessed_incoming_frames( + exec_ctx, &s->data_parser, s, + &s->unprocessed_incoming_frames_buffer, NULL, s->recv_message); + if (error != GRPC_ERROR_NONE) { + s->seen_error = true; + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, + &s->frame_storage); + grpc_slice_buffer_reset_and_unref_internal( + exec_ctx, &s->unprocessed_incoming_frames_buffer); + break; + } else if (*s->recv_message != NULL) { + break; + } + } + } + if (error == GRPC_ERROR_NONE && *s->recv_message != NULL) { null_then_run_closure(exec_ctx, &s->recv_message_ready, GRPC_ERROR_NONE); } else if (s->published_metadata[1] != GRPC_METADATA_NOT_PUBLISHED) { *s->recv_message = NULL; null_then_run_closure(exec_ctx, &s->recv_message_ready, GRPC_ERROR_NONE); } + GRPC_ERROR_UNREF(error); } } void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s) { - grpc_byte_stream *bs; grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s); if (s->recv_trailing_metadata_finished != NULL && s->read_closed && s->write_closed) { if (s->seen_error) { - while ((bs = grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames)) != - NULL) { - incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &s->frame_storage); + if (!s->pending_byte_stream) { + grpc_slice_buffer_reset_and_unref_internal( + exec_ctx, &s->unprocessed_incoming_frames_buffer); } } - if (s->all_incoming_byte_streams_finished && + bool pending_data = s->pending_byte_stream || + s->unprocessed_incoming_frames_buffer.length > 0; + if (s->read_closed && s->frame_storage.length == 0 && + (!pending_data || s->seen_error) && s->recv_trailing_metadata_finished != NULL) { grpc_chttp2_incoming_metadata_buffer_publish( exec_ctx, &s->metadata_buffer[1], s->recv_trailing_metadata); @@ -1677,14 +1731,6 @@ void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_exec_ctx *exec_ctx, } } -static void decrement_active_streams_locked(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t, - grpc_chttp2_stream *s) { - if ((s->all_incoming_byte_streams_finished = gpr_unref(&s->active_streams))) { - grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, t, s); - } -} - static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, uint32_t id, grpc_error *error) { grpc_chttp2_stream *s = grpc_chttp2_stream_map_delete(&t->stream_map, id); @@ -1693,10 +1739,19 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, t->incoming_stream = NULL; grpc_chttp2_parsing_become_skip_parser(exec_ctx, t); } - if (s->data_parser.parsing_frame != NULL) { - grpc_chttp2_incoming_byte_stream_finished( - exec_ctx, s->data_parser.parsing_frame, GRPC_ERROR_REF(error)); - s->data_parser.parsing_frame = NULL; + if (s->pending_byte_stream) { + if (s->on_next != NULL) { + grpc_chttp2_incoming_byte_stream *bs = s->data_parser.parsing_frame; + if (error == GRPC_ERROR_NONE) { + error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); + } + incoming_byte_stream_publish_error(exec_ctx, bs, error); + incoming_byte_stream_unref(exec_ctx, bs); + s->data_parser.parsing_frame = NULL; + } else { + GRPC_ERROR_UNREF(s->byte_stream_error); + s->byte_stream_error = GRPC_ERROR_REF(error); + } } if (grpc_chttp2_stream_map_size(&t->stream_map) == 0) { @@ -1882,7 +1937,6 @@ void grpc_chttp2_mark_stream_closed(grpc_exec_ctx *exec_ctx, s->published_metadata[i] = GPRC_METADATA_PUBLISHED_AT_CLOSE; } } - decrement_active_streams_locked(exec_ctx, t, s); grpc_chttp2_maybe_complete_recv_initial_metadata(exec_ctx, t, s); grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s); } @@ -2422,12 +2476,28 @@ static void set_pollset_set(grpc_exec_ctx *exec_ctx, grpc_transport *gt, * BYTE STREAM */ +static void reset_byte_stream(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_chttp2_stream *s = (grpc_chttp2_stream *)arg; + + s->pending_byte_stream = false; + if (error == GRPC_ERROR_NONE) { + grpc_chttp2_maybe_complete_recv_message(exec_ctx, s->t, s); + grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, s->t, s); + } else { + GPR_ASSERT(error != GRPC_ERROR_NONE); + grpc_closure_sched(exec_ctx, s->on_next, GRPC_ERROR_REF(error)); + s->on_next = NULL; + GRPC_ERROR_UNREF(s->byte_stream_error); + s->byte_stream_error = GRPC_ERROR_NONE; + grpc_chttp2_cancel_stream(exec_ctx, s->t, s, GRPC_ERROR_REF(error)); + s->byte_stream_error = error; + } +} + static void incoming_byte_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs) { if (gpr_unref(&bs->refs)) { - GRPC_ERROR_UNREF(bs->error); - grpc_slice_buffer_destroy_internal(exec_ctx, &bs->slices); - gpr_mu_destroy(&bs->slice_mu); gpr_free(bs); } } @@ -2487,47 +2557,90 @@ static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t = bs->transport; grpc_chttp2_stream *s = bs->stream; - if (bs->is_tail) { - gpr_mu_lock(&bs->slice_mu); - size_t cur_length = bs->slices.length; - gpr_mu_unlock(&bs->slice_mu); - incoming_byte_stream_update_flow_control( - exec_ctx, t, s, bs->next_action.max_size_hint, cur_length); - } - gpr_mu_lock(&bs->slice_mu); - if (bs->slices.count > 0) { - *bs->next_action.slice = grpc_slice_buffer_take_first(&bs->slices); - grpc_closure_run(exec_ctx, bs->next_action.on_complete, GRPC_ERROR_NONE); - } else if (bs->error != GRPC_ERROR_NONE) { - grpc_closure_run(exec_ctx, bs->next_action.on_complete, - GRPC_ERROR_REF(bs->error)); + size_t cur_length = s->frame_storage.length; + incoming_byte_stream_update_flow_control( + exec_ctx, t, s, bs->next_action.max_size_hint, cur_length); + + GPR_ASSERT(s->unprocessed_incoming_frames_buffer.length == 0); + if (s->frame_storage.length > 0) { + grpc_slice_buffer_swap(&s->frame_storage, + &s->unprocessed_incoming_frames_buffer); + grpc_closure_sched(exec_ctx, bs->next_action.on_complete, GRPC_ERROR_NONE); + } else if (s->byte_stream_error != GRPC_ERROR_NONE) { + grpc_closure_sched(exec_ctx, bs->next_action.on_complete, + GRPC_ERROR_REF(s->byte_stream_error)); + if (s->data_parser.parsing_frame != NULL) { + incoming_byte_stream_unref(exec_ctx, s->data_parser.parsing_frame); + s->data_parser.parsing_frame = NULL; + } + } else if (s->read_closed) { + if (bs->remaining_bytes != 0) { + s->byte_stream_error = + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); + grpc_closure_sched(exec_ctx, bs->next_action.on_complete, + GRPC_ERROR_REF(s->byte_stream_error)); + if (s->data_parser.parsing_frame != NULL) { + incoming_byte_stream_unref(exec_ctx, s->data_parser.parsing_frame); + s->data_parser.parsing_frame = NULL; + } + } else { + /* Should never reach here. */ + GPR_ASSERT(false); + } } else { - bs->on_next = bs->next_action.on_complete; - bs->next = bs->next_action.slice; + s->on_next = bs->next_action.on_complete; } - gpr_mu_unlock(&bs->slice_mu); incoming_byte_stream_unref(exec_ctx, bs); } -static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx, - grpc_byte_stream *byte_stream, - grpc_slice *slice, size_t max_size_hint, - grpc_closure *on_complete) { +static bool incoming_byte_stream_next(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + size_t max_size_hint, + grpc_closure *on_complete) { GPR_TIMER_BEGIN("incoming_byte_stream_next", 0); grpc_chttp2_incoming_byte_stream *bs = (grpc_chttp2_incoming_byte_stream *)byte_stream; - gpr_ref(&bs->refs); - bs->next_action.slice = slice; - bs->next_action.max_size_hint = max_size_hint; - bs->next_action.on_complete = on_complete; - grpc_closure_sched( - exec_ctx, - grpc_closure_init( - &bs->next_action.closure, incoming_byte_stream_next_locked, bs, - grpc_combiner_scheduler(bs->transport->combiner, false)), - GRPC_ERROR_NONE); - GPR_TIMER_END("incoming_byte_stream_next", 0); - return 0; + grpc_chttp2_stream *s = bs->stream; + if (s->unprocessed_incoming_frames_buffer.length > 0) { + return true; + } else { + gpr_ref(&bs->refs); + bs->next_action.max_size_hint = max_size_hint; + bs->next_action.on_complete = on_complete; + grpc_closure_sched( + exec_ctx, + grpc_closure_init( + &bs->next_action.closure, incoming_byte_stream_next_locked, bs, + grpc_combiner_scheduler(bs->transport->combiner, false)), + GRPC_ERROR_NONE); + GPR_TIMER_END("incoming_byte_stream_next", 0); + return false; + } +} + +static grpc_error *incoming_byte_stream_pull(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + grpc_slice *slice) { + GPR_TIMER_BEGIN("incoming_byte_stream_pull", 0); + grpc_chttp2_incoming_byte_stream *bs = + (grpc_chttp2_incoming_byte_stream *)byte_stream; + grpc_chttp2_stream *s = bs->stream; + + if (s->unprocessed_incoming_frames_buffer.length > 0) { + grpc_error *error = grpc_deframe_unprocessed_incoming_frames( + exec_ctx, &s->data_parser, s, &s->unprocessed_incoming_frames_buffer, + slice, NULL); + if (error != GRPC_ERROR_NONE) { + return error; + } + } else { + grpc_error *error = + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); + grpc_closure_sched(exec_ctx, &s->reset_byte_stream, GRPC_ERROR_REF(error)); + return error; + } + GPR_TIMER_END("incoming_byte_stream_pull", 0); + return GRPC_ERROR_NONE; } static void incoming_byte_stream_destroy(grpc_exec_ctx *exec_ctx, @@ -2537,9 +2650,14 @@ static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx, void *byte_stream, grpc_error *error_ignored) { grpc_chttp2_incoming_byte_stream *bs = byte_stream; + grpc_chttp2_stream *s = bs->stream; + grpc_chttp2_transport *t = s->t; + GPR_ASSERT(bs->base.destroy == incoming_byte_stream_destroy); - decrement_active_streams_locked(exec_ctx, bs->transport, bs->stream); incoming_byte_stream_unref(exec_ctx, bs); + s->pending_byte_stream = false; + grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s); + grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, t, s); } static void incoming_byte_stream_destroy(grpc_exec_ctx *exec_ctx, @@ -2559,50 +2677,53 @@ static void incoming_byte_stream_destroy(grpc_exec_ctx *exec_ctx, static void incoming_byte_stream_publish_error( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs, grpc_error *error) { + grpc_chttp2_stream *s = bs->stream; + GPR_ASSERT(error != GRPC_ERROR_NONE); - grpc_closure_sched(exec_ctx, bs->on_next, GRPC_ERROR_REF(error)); - bs->on_next = NULL; - GRPC_ERROR_UNREF(bs->error); + grpc_closure_sched(exec_ctx, s->on_next, GRPC_ERROR_REF(error)); + s->on_next = NULL; + GRPC_ERROR_UNREF(s->byte_stream_error); + s->byte_stream_error = GRPC_ERROR_REF(error); grpc_chttp2_cancel_stream(exec_ctx, bs->transport, bs->stream, GRPC_ERROR_REF(error)); - bs->error = error; } -void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx, - grpc_chttp2_incoming_byte_stream *bs, - grpc_slice slice) { - gpr_mu_lock(&bs->slice_mu); +grpc_error *grpc_chttp2_incoming_byte_stream_push( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs, + grpc_slice slice, grpc_slice *slice_out) { + grpc_chttp2_stream *s = bs->stream; + if (bs->remaining_bytes < GRPC_SLICE_LENGTH(slice)) { - incoming_byte_stream_publish_error( - exec_ctx, bs, - GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many bytes in stream")); + grpc_error *error = + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many bytes in stream"); + + grpc_closure_sched(exec_ctx, &s->reset_byte_stream, GRPC_ERROR_REF(error)); + grpc_slice_unref_internal(exec_ctx, slice); + return error; } else { bs->remaining_bytes -= (uint32_t)GRPC_SLICE_LENGTH(slice); - if (bs->on_next != NULL) { - *bs->next = slice; - grpc_closure_sched(exec_ctx, bs->on_next, GRPC_ERROR_NONE); - bs->on_next = NULL; - } else { - grpc_slice_buffer_add(&bs->slices, slice); + if (slice_out != NULL) { + *slice_out = slice; } + return GRPC_ERROR_NONE; } - gpr_mu_unlock(&bs->slice_mu); } -void grpc_chttp2_incoming_byte_stream_finished( +grpc_error *grpc_chttp2_incoming_byte_stream_finished( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs, - grpc_error *error) { + grpc_error *error, bool reset_on_error) { + grpc_chttp2_stream *s = bs->stream; + if (error == GRPC_ERROR_NONE) { - gpr_mu_lock(&bs->slice_mu); if (bs->remaining_bytes != 0) { error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); } - gpr_mu_unlock(&bs->slice_mu); } - if (error != GRPC_ERROR_NONE) { - incoming_byte_stream_publish_error(exec_ctx, bs, error); + if (error != GRPC_ERROR_NONE && reset_on_error) { + grpc_closure_sched(exec_ctx, &s->reset_byte_stream, GRPC_ERROR_REF(error)); } incoming_byte_stream_unref(exec_ctx, bs); + return error; } grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create( @@ -2614,26 +2735,12 @@ grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create( incoming_byte_stream->remaining_bytes = frame_size; incoming_byte_stream->base.flags = flags; incoming_byte_stream->base.next = incoming_byte_stream_next; + incoming_byte_stream->base.pull = incoming_byte_stream_pull; incoming_byte_stream->base.destroy = incoming_byte_stream_destroy; - gpr_mu_init(&incoming_byte_stream->slice_mu); gpr_ref_init(&incoming_byte_stream->refs, 2); - incoming_byte_stream->next_message = NULL; incoming_byte_stream->transport = t; incoming_byte_stream->stream = s; - gpr_ref(&incoming_byte_stream->stream->active_streams); - grpc_slice_buffer_init(&incoming_byte_stream->slices); - incoming_byte_stream->on_next = NULL; - incoming_byte_stream->is_tail = 1; - incoming_byte_stream->error = GRPC_ERROR_NONE; - grpc_chttp2_incoming_frame_queue *q = &s->incoming_frames; - if (q->head == NULL) { - q->head = incoming_byte_stream; - } else { - q->tail->is_tail = 0; - q->tail->next_message = incoming_byte_stream; - } - q->tail = incoming_byte_stream; - grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s); + s->byte_stream_error = GRPC_ERROR_NONE; return incoming_byte_stream; } diff --git a/src/core/ext/transport/chttp2/transport/frame_data.c b/src/core/ext/transport/chttp2/transport/frame_data.c index 6e9258ee7e..9aba3ea445 100644 --- a/src/core/ext/transport/chttp2/transport/frame_data.c +++ b/src/core/ext/transport/chttp2/transport/frame_data.c @@ -40,6 +40,7 @@ #include <grpc/support/string_util.h> #include <grpc/support/useful.h> #include "src/core/ext/transport/chttp2/transport/internal.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/transport.h" @@ -53,16 +54,17 @@ grpc_error *grpc_chttp2_data_parser_init(grpc_chttp2_data_parser *parser) { void grpc_chttp2_data_parser_destroy(grpc_exec_ctx *exec_ctx, grpc_chttp2_data_parser *parser) { if (parser->parsing_frame != NULL) { - grpc_chttp2_incoming_byte_stream_finished( + GRPC_ERROR_UNREF(grpc_chttp2_incoming_byte_stream_finished( exec_ctx, parser->parsing_frame, - GRPC_ERROR_CREATE_FROM_STATIC_STRING("Parser destroyed")); + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Parser destroyed"), false)); } GRPC_ERROR_UNREF(parser->error); } grpc_error *grpc_chttp2_data_parser_begin_frame(grpc_chttp2_data_parser *parser, uint8_t flags, - uint32_t stream_id) { + uint32_t stream_id, + grpc_chttp2_stream *s) { if (flags & ~GRPC_CHTTP2_DATA_FLAG_END_STREAM) { char *msg; gpr_asprintf(&msg, "unsupported data flags: 0x%02x", flags); @@ -74,47 +76,14 @@ grpc_error *grpc_chttp2_data_parser_begin_frame(grpc_chttp2_data_parser *parser, } if (flags & GRPC_CHTTP2_DATA_FLAG_END_STREAM) { - parser->is_last_frame = 1; + s->received_last_frame = true; } else { - parser->is_last_frame = 0; + s->received_last_frame = false; } return GRPC_ERROR_NONE; } -void grpc_chttp2_incoming_frame_queue_merge( - grpc_chttp2_incoming_frame_queue *head_dst, - grpc_chttp2_incoming_frame_queue *tail_src) { - if (tail_src->head == NULL) { - return; - } - - if (head_dst->head == NULL) { - *head_dst = *tail_src; - memset(tail_src, 0, sizeof(*tail_src)); - return; - } - - head_dst->tail->next_message = tail_src->head; - head_dst->tail = tail_src->tail; - memset(tail_src, 0, sizeof(*tail_src)); -} - -grpc_byte_stream *grpc_chttp2_incoming_frame_queue_pop( - grpc_chttp2_incoming_frame_queue *q) { - grpc_byte_stream *out; - if (q->head == NULL) { - return NULL; - } - out = &q->head->base; - if (q->head == q->tail) { - memset(q, 0, sizeof(*q)); - } else { - q->head = q->head->next_message; - } - return out; -} - void grpc_chttp2_encode_data(uint32_t id, grpc_slice_buffer *inbuf, uint32_t write_bytes, int is_eof, grpc_transport_one_way_stats *stats, @@ -143,145 +112,215 @@ void grpc_chttp2_encode_data(uint32_t id, grpc_slice_buffer *inbuf, stats->data_bytes += write_bytes; } -static grpc_error *parse_inner(grpc_exec_ctx *exec_ctx, - grpc_chttp2_data_parser *p, - grpc_chttp2_transport *t, grpc_chttp2_stream *s, - grpc_slice slice) { - uint8_t *const beg = GRPC_SLICE_START_PTR(slice); - uint8_t *const end = GRPC_SLICE_END_PTR(slice); - uint8_t *cur = beg; - uint32_t message_flags; - grpc_chttp2_incoming_byte_stream *incoming_byte_stream; - char *msg; +grpc_error *grpc_deframe_unprocessed_incoming_frames( + grpc_exec_ctx *exec_ctx, grpc_chttp2_data_parser *p, grpc_chttp2_stream *s, + grpc_slice_buffer *slices, grpc_slice *slice_out, + grpc_byte_stream **stream_out) { + grpc_error *error = GRPC_ERROR_NONE; + grpc_chttp2_transport *t = s->t; - if (cur == end) { - return GRPC_ERROR_NONE; - } + while (slices->count > 0) { + uint8_t *beg = NULL; + uint8_t *end = NULL; + uint8_t *cur = NULL; - switch (p->state) { - case GRPC_CHTTP2_DATA_ERROR: - p->state = GRPC_CHTTP2_DATA_ERROR; - return GRPC_ERROR_REF(p->error); - fh_0: - case GRPC_CHTTP2_DATA_FH_0: - s->stats.incoming.framing_bytes++; - p->frame_type = *cur; - switch (p->frame_type) { - case 0: - p->is_frame_compressed = 0; /* GPR_FALSE */ - break; - case 1: - p->is_frame_compressed = 1; /* GPR_TRUE */ - break; - default: - gpr_asprintf(&msg, "Bad GRPC frame type 0x%02x", p->frame_type); - p->error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); - p->error = grpc_error_set_int(p->error, GRPC_ERROR_INT_STREAM_ID, - (intptr_t)s->id); - gpr_free(msg); - msg = grpc_dump_slice(slice, GPR_DUMP_HEX | GPR_DUMP_ASCII); - p->error = grpc_error_set_str(p->error, GRPC_ERROR_STR_RAW_BYTES, - grpc_slice_from_copied_string(msg)); - gpr_free(msg); - p->error = - grpc_error_set_int(p->error, GRPC_ERROR_INT_OFFSET, cur - beg); - p->state = GRPC_CHTTP2_DATA_ERROR; - return GRPC_ERROR_REF(p->error); - } - if (++cur == end) { - p->state = GRPC_CHTTP2_DATA_FH_1; - return GRPC_ERROR_NONE; - } - /* fallthrough */ - case GRPC_CHTTP2_DATA_FH_1: - s->stats.incoming.framing_bytes++; - p->frame_size = ((uint32_t)*cur) << 24; - if (++cur == end) { - p->state = GRPC_CHTTP2_DATA_FH_2; - return GRPC_ERROR_NONE; - } - /* fallthrough */ - case GRPC_CHTTP2_DATA_FH_2: - s->stats.incoming.framing_bytes++; - p->frame_size |= ((uint32_t)*cur) << 16; - if (++cur == end) { - p->state = GRPC_CHTTP2_DATA_FH_3; - return GRPC_ERROR_NONE; - } - /* fallthrough */ - case GRPC_CHTTP2_DATA_FH_3: - s->stats.incoming.framing_bytes++; - p->frame_size |= ((uint32_t)*cur) << 8; - if (++cur == end) { - p->state = GRPC_CHTTP2_DATA_FH_4; - return GRPC_ERROR_NONE; - } - /* fallthrough */ - case GRPC_CHTTP2_DATA_FH_4: - s->stats.incoming.framing_bytes++; - p->frame_size |= ((uint32_t)*cur); - p->state = GRPC_CHTTP2_DATA_FRAME; - ++cur; - message_flags = 0; - if (p->is_frame_compressed) { - message_flags |= GRPC_WRITE_INTERNAL_COMPRESS; - } - p->parsing_frame = incoming_byte_stream = - grpc_chttp2_incoming_byte_stream_create(exec_ctx, t, s, p->frame_size, - message_flags); - /* fallthrough */ - case GRPC_CHTTP2_DATA_FRAME: - if (cur == end) { - return GRPC_ERROR_NONE; - } - uint32_t remaining = (uint32_t)(end - cur); - if (remaining == p->frame_size) { - s->stats.incoming.data_bytes += p->frame_size; - grpc_chttp2_incoming_byte_stream_push( - exec_ctx, p->parsing_frame, - grpc_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); - grpc_chttp2_incoming_byte_stream_finished(exec_ctx, p->parsing_frame, - GRPC_ERROR_NONE); - p->parsing_frame = NULL; - p->state = GRPC_CHTTP2_DATA_FH_0; - return GRPC_ERROR_NONE; - } else if (remaining > p->frame_size) { - s->stats.incoming.data_bytes += p->frame_size; - grpc_chttp2_incoming_byte_stream_push( - exec_ctx, p->parsing_frame, - grpc_slice_sub(slice, (size_t)(cur - beg), - (size_t)(cur + p->frame_size - beg))); - grpc_chttp2_incoming_byte_stream_finished(exec_ctx, p->parsing_frame, - GRPC_ERROR_NONE); - p->parsing_frame = NULL; - cur += p->frame_size; - goto fh_0; /* loop */ - } else { - GPR_ASSERT(remaining <= p->frame_size); - grpc_chttp2_incoming_byte_stream_push( - exec_ctx, p->parsing_frame, - grpc_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); - p->frame_size -= remaining; - s->stats.incoming.data_bytes += remaining; + grpc_slice slice = grpc_slice_buffer_take_first(slices); + + beg = GRPC_SLICE_START_PTR(slice); + end = GRPC_SLICE_END_PTR(slice); + cur = beg; + uint32_t message_flags; + char *msg; + + if (cur == end) { + grpc_slice_unref_internal(exec_ctx, slice); + continue; + } + + switch (p->state) { + case GRPC_CHTTP2_DATA_ERROR: + p->state = GRPC_CHTTP2_DATA_ERROR; + grpc_slice_unref_internal(exec_ctx, slice); + return GRPC_ERROR_REF(p->error); + case GRPC_CHTTP2_DATA_FH_0: + p->frame_type = *cur; + switch (p->frame_type) { + case 0: + p->is_frame_compressed = false; /* GPR_FALSE */ + break; + case 1: + p->is_frame_compressed = true; /* GPR_TRUE */ + break; + default: + gpr_asprintf(&msg, "Bad GRPC frame type 0x%02x", p->frame_type); + p->error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); + p->error = grpc_error_set_int(p->error, GRPC_ERROR_INT_STREAM_ID, + (intptr_t)s->id); + gpr_free(msg); + msg = grpc_dump_slice(slice, GPR_DUMP_HEX | GPR_DUMP_ASCII); + p->error = grpc_error_set_str(p->error, GRPC_ERROR_STR_RAW_BYTES, + grpc_slice_from_copied_string(msg)); + gpr_free(msg); + p->error = + grpc_error_set_int(p->error, GRPC_ERROR_INT_OFFSET, cur - beg); + p->state = GRPC_CHTTP2_DATA_ERROR; + grpc_slice_unref_internal(exec_ctx, slice); + return GRPC_ERROR_REF(p->error); + } + if (++cur == end) { + p->state = GRPC_CHTTP2_DATA_FH_1; + grpc_slice_unref_internal(exec_ctx, slice); + continue; + } + /* fallthrough */ + case GRPC_CHTTP2_DATA_FH_1: + p->frame_size = ((uint32_t)*cur) << 24; + if (++cur == end) { + p->state = GRPC_CHTTP2_DATA_FH_2; + grpc_slice_unref_internal(exec_ctx, slice); + continue; + } + /* fallthrough */ + case GRPC_CHTTP2_DATA_FH_2: + p->frame_size |= ((uint32_t)*cur) << 16; + if (++cur == end) { + p->state = GRPC_CHTTP2_DATA_FH_3; + grpc_slice_unref_internal(exec_ctx, slice); + continue; + } + /* fallthrough */ + case GRPC_CHTTP2_DATA_FH_3: + p->frame_size |= ((uint32_t)*cur) << 8; + if (++cur == end) { + p->state = GRPC_CHTTP2_DATA_FH_4; + grpc_slice_unref_internal(exec_ctx, slice); + continue; + } + /* fallthrough */ + case GRPC_CHTTP2_DATA_FH_4: + GPR_ASSERT(stream_out != NULL); + GPR_ASSERT(p->parsing_frame == NULL); + p->frame_size |= ((uint32_t)*cur); + p->state = GRPC_CHTTP2_DATA_FRAME; + ++cur; + message_flags = 0; + if (p->is_frame_compressed) { + message_flags |= GRPC_WRITE_INTERNAL_COMPRESS; + } + p->parsing_frame = grpc_chttp2_incoming_byte_stream_create( + exec_ctx, t, s, p->frame_size, message_flags); + *stream_out = &p->parsing_frame->base; + if (p->parsing_frame->remaining_bytes == 0) { + GRPC_ERROR_UNREF(grpc_chttp2_incoming_byte_stream_finished( + exec_ctx, p->parsing_frame, GRPC_ERROR_NONE, true)); + p->parsing_frame = NULL; + p->state = GRPC_CHTTP2_DATA_FH_0; + } + s->pending_byte_stream = true; + + if (cur != end) { + grpc_slice_buffer_undo_take_first( + &s->unprocessed_incoming_frames_buffer, + grpc_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); + } + grpc_slice_unref_internal(exec_ctx, slice); return GRPC_ERROR_NONE; + case GRPC_CHTTP2_DATA_FRAME: { + GPR_ASSERT(p->parsing_frame != NULL); + GPR_ASSERT(slice_out != NULL); + if (cur == end) { + grpc_slice_unref_internal(exec_ctx, slice); + continue; + } + uint32_t remaining = (uint32_t)(end - cur); + if (remaining == p->frame_size) { + if (GRPC_ERROR_NONE != (error = grpc_chttp2_incoming_byte_stream_push( + exec_ctx, p->parsing_frame, + grpc_slice_sub(slice, (size_t)(cur - beg), + (size_t)(end - beg)), + slice_out))) { + grpc_slice_unref_internal(exec_ctx, slice); + return error; + } + if (GRPC_ERROR_NONE != + (error = grpc_chttp2_incoming_byte_stream_finished( + exec_ctx, p->parsing_frame, GRPC_ERROR_NONE, true))) { + grpc_slice_unref_internal(exec_ctx, slice); + return error; + } + p->parsing_frame = NULL; + p->state = GRPC_CHTTP2_DATA_FH_0; + grpc_slice_unref_internal(exec_ctx, slice); + return GRPC_ERROR_NONE; + } else if (remaining < p->frame_size) { + if (GRPC_ERROR_NONE != (error = grpc_chttp2_incoming_byte_stream_push( + exec_ctx, p->parsing_frame, + grpc_slice_sub(slice, (size_t)(cur - beg), + (size_t)(end - beg)), + slice_out))) { + return error; + } + p->frame_size -= remaining; + grpc_slice_unref_internal(exec_ctx, slice); + return GRPC_ERROR_NONE; + } else { + GPR_ASSERT(remaining > p->frame_size); + if (GRPC_ERROR_NONE != + (grpc_chttp2_incoming_byte_stream_push( + exec_ctx, p->parsing_frame, + grpc_slice_sub(slice, (size_t)(cur - beg), + (size_t)(cur + p->frame_size - beg)), + slice_out))) { + grpc_slice_unref_internal(exec_ctx, slice); + return error; + } + if (GRPC_ERROR_NONE != + (error = grpc_chttp2_incoming_byte_stream_finished( + exec_ctx, p->parsing_frame, GRPC_ERROR_NONE, true))) { + grpc_slice_unref_internal(exec_ctx, slice); + return error; + } + p->parsing_frame = NULL; + p->state = GRPC_CHTTP2_DATA_FH_0; + cur += p->frame_size; + grpc_slice_buffer_undo_take_first( + &s->unprocessed_incoming_frames_buffer, + grpc_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); + grpc_slice_unref_internal(exec_ctx, slice); + return GRPC_ERROR_NONE; + } } + } } - GPR_UNREACHABLE_CODE( - return GRPC_ERROR_CREATE_FROM_STATIC_STRING("Should never reach here")); + return GRPC_ERROR_NONE; } grpc_error *grpc_chttp2_data_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_slice slice, int is_last) { - grpc_chttp2_data_parser *p = parser; - grpc_error *error = parse_inner(exec_ctx, p, t, s, slice); + /* grpc_error *error = parse_inner_buffer(exec_ctx, p, t, s, slice); */ + s->stats.incoming.framing_bytes += GRPC_SLICE_LENGTH(slice); + if (!s->pending_byte_stream) { + grpc_slice_ref_internal(slice); + grpc_slice_buffer_add(&s->frame_storage, slice); + grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s); + } else if (s->on_next) { + GPR_ASSERT(s->frame_storage.length == 0); + grpc_slice_ref_internal(slice); + grpc_slice_buffer_add(&s->unprocessed_incoming_frames_buffer, slice); + grpc_closure_sched(exec_ctx, s->on_next, GRPC_ERROR_NONE); + s->on_next = NULL; + } else { + grpc_slice_ref_internal(slice); + grpc_slice_buffer_add(&s->frame_storage, slice); + } - if (is_last && p->is_last_frame) { + if (is_last && s->received_last_frame) { grpc_chttp2_mark_stream_closed(exec_ctx, t, s, true, false, GRPC_ERROR_NONE); } - return error; + return GRPC_ERROR_NONE; } diff --git a/src/core/ext/transport/chttp2/transport/frame_data.h b/src/core/ext/transport/chttp2/transport/frame_data.h index 264ad14608..9ed4ad0f21 100644 --- a/src/core/ext/transport/chttp2/transport/frame_data.h +++ b/src/core/ext/transport/chttp2/transport/frame_data.h @@ -56,28 +56,16 @@ typedef enum { typedef struct grpc_chttp2_incoming_byte_stream grpc_chttp2_incoming_byte_stream; -typedef struct grpc_chttp2_incoming_frame_queue { - grpc_chttp2_incoming_byte_stream *head; - grpc_chttp2_incoming_byte_stream *tail; -} grpc_chttp2_incoming_frame_queue; - typedef struct { grpc_chttp2_stream_state state; - uint8_t is_last_frame; uint8_t frame_type; uint32_t frame_size; grpc_error *error; - int is_frame_compressed; + bool is_frame_compressed; grpc_chttp2_incoming_byte_stream *parsing_frame; } grpc_chttp2_data_parser; -void grpc_chttp2_incoming_frame_queue_merge( - grpc_chttp2_incoming_frame_queue *head_dst, - grpc_chttp2_incoming_frame_queue *tail_src); -grpc_byte_stream *grpc_chttp2_incoming_frame_queue_pop( - grpc_chttp2_incoming_frame_queue *q); - /* initialize per-stream state for data frame parsing */ grpc_error *grpc_chttp2_data_parser_init(grpc_chttp2_data_parser *parser); @@ -87,7 +75,8 @@ void grpc_chttp2_data_parser_destroy(grpc_exec_ctx *exec_ctx, /* start processing a new data frame */ grpc_error *grpc_chttp2_data_parser_begin_frame(grpc_chttp2_data_parser *parser, uint8_t flags, - uint32_t stream_id); + uint32_t stream_id, + grpc_chttp2_stream *s); /* handle a slice of a data frame - is_last indicates the last slice of a frame */ @@ -101,4 +90,9 @@ void grpc_chttp2_encode_data(uint32_t id, grpc_slice_buffer *inbuf, grpc_transport_one_way_stats *stats, grpc_slice_buffer *outbuf); +grpc_error *grpc_deframe_unprocessed_incoming_frames( + grpc_exec_ctx *exec_ctx, grpc_chttp2_data_parser *p, grpc_chttp2_stream *s, + grpc_slice_buffer *slices, grpc_slice *slice_out, + grpc_byte_stream **stream_out); + #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_DATA_H */ diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 0452081dd5..0aaa4aebe5 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -195,22 +195,20 @@ typedef struct grpc_chttp2_write_cb { struct grpc_chttp2_incoming_byte_stream { grpc_byte_stream base; gpr_refcount refs; - struct grpc_chttp2_incoming_byte_stream *next_message; - grpc_error *error; - grpc_chttp2_transport *transport; - grpc_chttp2_stream *stream; - bool is_tail; + grpc_chttp2_transport *transport; /* immutable */ + grpc_chttp2_stream *stream; /* immutable */ - gpr_mu slice_mu; // protects slices, on_next - grpc_slice_buffer slices; - grpc_closure *on_next; - grpc_slice *next; + /* Accessed only by transport thread when stream->pending_byte_stream == false + * Accessed only by application thread when stream->pending_byte_stream == + * true */ uint32_t remaining_bytes; + /* Accessed only by transport thread when stream->pending_byte_stream == false + * Accessed only by application thread when stream->pending_byte_stream == + * true */ struct { grpc_closure closure; - grpc_slice *slice; size_t max_size_hint; grpc_closure *on_complete; } next_action; @@ -446,8 +444,8 @@ struct grpc_chttp2_stream { uint32_t id; /** window available for us to send to peer, over or under the initial window - * size of the transport... ie: - * outgoing_window = outgoing_window_delta + transport.initial_window_size */ + * size of the transport... ie: + * outgoing_window = outgoing_window_delta + transport.initial_window_size */ int64_t outgoing_window_delta; /** things the upper layers would like to send */ grpc_metadata_batch *send_initial_metadata; @@ -474,9 +472,6 @@ struct grpc_chttp2_stream { grpc_transport_stream_stats *collecting_stats; grpc_transport_stream_stats stats; - /** number of streams that are currently being read */ - gpr_refcount active_streams; - /** Is this stream closed for writing. */ bool write_closed; /** Is this stream reading half-closed. */ @@ -500,7 +495,17 @@ struct grpc_chttp2_stream { grpc_chttp2_incoming_metadata_buffer metadata_buffer[2]; - grpc_chttp2_incoming_frame_queue incoming_frames; + grpc_slice_buffer frame_storage; /* protected by t combiner */ + + /* Accessed only by transport thread when stream->pending_byte_stream == false + * Accessed only by application thread when stream->pending_byte_stream == + * true */ + grpc_slice_buffer unprocessed_incoming_frames_buffer; + grpc_closure *on_next; /* protected by t combiner */ + bool pending_byte_stream; /* protected by t combiner */ + grpc_closure reset_byte_stream; + grpc_error *byte_stream_error; /* protected by t combiner */ + bool received_last_frame; /* protected by t combiner */ gpr_timespec deadline; @@ -513,6 +518,9 @@ struct grpc_chttp2_stream { * incoming_window = incoming_window_delta + transport.initial_window_size */ int64_t incoming_window_delta; /** parsing state for data frames */ + /* Accessed only by transport thread when stream->pending_byte_stream == false + * Accessed only by application thread when stream->pending_byte_stream == + * true */ grpc_chttp2_data_parser data_parser; /** number of bytes received - reset at end of parse thread execution */ int64_t received_bytes; @@ -791,10 +799,13 @@ void grpc_chttp2_ref_transport(grpc_chttp2_transport *t); grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, uint32_t frame_size, uint32_t flags); -void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx, - grpc_chttp2_incoming_byte_stream *bs, - grpc_slice slice); -void grpc_chttp2_incoming_byte_stream_finished( +grpc_error *grpc_chttp2_incoming_byte_stream_push( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs, + grpc_slice slice, grpc_slice *slice_out); +grpc_error *grpc_chttp2_incoming_byte_stream_finished( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs, + grpc_error *error, bool reset_on_error); +void grpc_chttp2_incoming_byte_stream_notify( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs, grpc_error *error); diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index 7e457ced27..638b137316 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -458,12 +458,13 @@ static grpc_error *init_data_frame_parser(grpc_exec_ctx *exec_ctx, return init_skip_frame_parser(exec_ctx, t, 0); } if (err == GRPC_ERROR_NONE) { - err = grpc_chttp2_data_parser_begin_frame(&s->data_parser, - t->incoming_frame_flags, s->id); + err = grpc_chttp2_data_parser_begin_frame( + &s->data_parser, t->incoming_frame_flags, s->id, s); } error_handler: if (err == GRPC_ERROR_NONE) { t->incoming_stream = s; + /* t->parser = grpc_chttp2_data_parser_parse;*/ t->parser = grpc_chttp2_data_parser_parse; t->parser_data = &s->data_parser; return GRPC_ERROR_NONE; diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c index 88335ecd0b..7896c70f9e 100644 --- a/src/core/ext/transport/cronet/transport/cronet_transport.c +++ b/src/core/ext/transport/cronet/transport/cronet_transport.c @@ -973,9 +973,20 @@ 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->payload->send_message.send_message, &slice, - stream_op->payload->send_message.send_message->length, NULL); + if (1 != grpc_byte_stream_next( + exec_ctx, stream_op->payload->send_message.send_message, + stream_op->payload->send_message.send_message->length, + NULL)) { + /* Should never reach here */ + GPR_ASSERT(false); + } + if (GRPC_ERROR_NONE != + grpc_byte_stream_pull(exec_ctx, + stream_op->payload->send_message.send_message, + &slice)) { + /* Should never reach here */ + GPR_ASSERT(false); + } grpc_slice_buffer_add(&write_slice_buffer, slice); if (write_slice_buffer.count != 1) { /* Empty request not handled yet */ diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c index 3de31d99da..238d176dfa 100644 --- a/src/core/lib/channel/channel_args.c +++ b/src/core/lib/channel/channel_args.c @@ -31,17 +31,18 @@ * */ -#include "src/core/lib/channel/channel_args.h" -#include <grpc/grpc.h> -#include "src/core/lib/support/string.h" +#include <limits.h> +#include <string.h> #include <grpc/compression.h> +#include <grpc/grpc.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/string_util.h> #include <grpc/support/useful.h> -#include <string.h> +#include "src/core/lib/channel/channel_args.h" +#include "src/core/lib/support/string.h" static grpc_arg copy_arg(const grpc_arg *src) { grpc_arg dst; @@ -330,7 +331,7 @@ const grpc_arg *grpc_channel_args_find(const grpc_channel_args *args, } int grpc_channel_arg_get_integer(const grpc_arg *arg, - grpc_integer_options options) { + const grpc_integer_options options) { if (arg == NULL) return options.default_value; if (arg->type != GRPC_ARG_INTEGER) { gpr_log(GPR_ERROR, "%s ignored: it must be an integer", arg->key); diff --git a/src/core/lib/channel/channel_args.h b/src/core/lib/channel/channel_args.h index 5ffcacb7fd..f0f603e251 100644 --- a/src/core/lib/channel/channel_args.h +++ b/src/core/lib/channel/channel_args.h @@ -120,9 +120,10 @@ typedef struct grpc_integer_options { int min_value; int max_value; } grpc_integer_options; + /** Returns the value of \a arg, subject to the contraints in \a options. */ int grpc_channel_arg_get_integer(const grpc_arg *arg, - grpc_integer_options options); + const grpc_integer_options options); bool grpc_channel_arg_get_bool(const grpc_arg *arg, bool default_value); diff --git a/src/core/lib/channel/context.h b/src/core/lib/channel/context.h index 2c1174ce7a..6c931ad28a 100644 --- a/src/core/lib/channel/context.h +++ b/src/core/lib/channel/context.h @@ -50,9 +50,6 @@ typedef enum { /// Reserved for traffic_class_context. GRPC_CONTEXT_TRAFFIC, - /// Costs for Load Reporting. - GRPC_CONTEXT_LR_COST, - GRPC_CONTEXT_COUNT } grpc_context_index; diff --git a/src/core/lib/iomgr/sockaddr_utils.h b/src/core/lib/iomgr/sockaddr_utils.h index 2b22f11b49..be3ea2038f 100644 --- a/src/core/lib/iomgr/sockaddr_utils.h +++ b/src/core/lib/iomgr/sockaddr_utils.h @@ -50,7 +50,7 @@ int grpc_sockaddr_to_v4mapped(const grpc_resolved_address *addr, grpc_resolved_address *addr6_out); /* If addr is ::, 0.0.0.0, or ::ffff:0.0.0.0, writes the port number to - *port_out (if not NULL) and returns true, otherwise returns false. */ + *port_out (if not NULL) and returns true, otherwise returns false. */ int grpc_sockaddr_is_wildcard(const grpc_resolved_address *addr, int *port_out); /* Writes 0.0.0.0:port and [::]:port to separate sockaddrs. */ diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index 60579e18ba..ca283d034f 100644 --- a/src/core/lib/iomgr/udp_server.c +++ b/src/core/lib/iomgr/udp_server.c @@ -79,10 +79,15 @@ struct grpc_udp_listener { grpc_resolved_address addr; grpc_closure read_closure; grpc_closure write_closure; + // To be called when corresponding QuicGrpcServer closes all active + // connections. + grpc_closure orphan_fd_closure; grpc_closure destroyed_closure; grpc_udp_server_read_cb read_cb; grpc_udp_server_write_cb write_cb; grpc_udp_server_orphan_cb orphan_cb; + // True if orphan_cb is trigered. + bool orphan_notified; struct grpc_udp_listener *next; }; @@ -146,6 +151,14 @@ grpc_udp_server *grpc_udp_server_create(const grpc_channel_args *args) { return s; } +static void shutdown_fd(grpc_exec_ctx *exec_ctx, void *fd, grpc_error *error) { + grpc_fd_shutdown(exec_ctx, (grpc_fd *)fd, GRPC_ERROR_REF(error)); +} + +static void dummy_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + // No-op. +} + static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { if (s->shutdown_complete != NULL) { grpc_closure_sched(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE); @@ -195,12 +208,16 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { grpc_closure_init(&sp->destroyed_closure, destroyed_port, s, grpc_schedule_on_exec_ctx); - - /* 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->server->user_data); - + if (!sp->orphan_notified) { + /* Call the orphan_cb to signal that the FD is about to be closed and + * should no longer be used. Because at this point, all listening ports + * have been shutdown already, no need to shutdown again.*/ + grpc_closure_init(&sp->orphan_fd_closure, dummy_cb, sp->emfd, + grpc_schedule_on_exec_ctx); + GPR_ASSERT(sp->orphan_cb); + sp->orphan_cb(exec_ctx, sp->emfd, &sp->orphan_fd_closure, + sp->server->user_data); + } grpc_fd_orphan(exec_ctx, sp->emfd, &sp->destroyed_closure, NULL, "udp_listener_shutdown"); } @@ -225,9 +242,11 @@ 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->server->user_data); - grpc_fd_shutdown(exec_ctx, sp->emfd, GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "Server destroyed")); + grpc_closure_init(&sp->orphan_fd_closure, shutdown_fd, sp->emfd, + grpc_schedule_on_exec_ctx); + sp->orphan_cb(exec_ctx, sp->emfd, &sp->orphan_fd_closure, + sp->server->user_data); + sp->orphan_notified = true; } gpr_mu_unlock(&s->mu); } else { @@ -391,6 +410,7 @@ static int add_socket_to_server(grpc_udp_server *s, int fd, sp->read_cb = read_cb; sp->write_cb = write_cb; sp->orphan_cb = orphan_cb; + sp->orphan_notified = false; GPR_ASSERT(sp->emfd); gpr_mu_unlock(&s->mu); gpr_free(name); diff --git a/src/core/lib/iomgr/udp_server.h b/src/core/lib/iomgr/udp_server.h index 9df3fe4d1f..8006037644 100644 --- a/src/core/lib/iomgr/udp_server.h +++ b/src/core/lib/iomgr/udp_server.h @@ -55,7 +55,9 @@ typedef void (*grpc_udp_server_write_cb)(grpc_exec_ctx *exec_ctx, grpc_fd *emfd, /* 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, void *user_data); + grpc_fd *emfd, + grpc_closure *shutdown_fd_callback, + void *user_data); /* Create a server, initially not bound to any ports */ grpc_udp_server *grpc_udp_server_create(const grpc_channel_args *args); diff --git a/src/core/lib/security/credentials/fake/fake_credentials.c b/src/core/lib/security/credentials/fake/fake_credentials.c index 68636ba208..3fdb67fb91 100644 --- a/src/core/lib/security/credentials/fake/fake_credentials.c +++ b/src/core/lib/security/credentials/fake/fake_credentials.c @@ -39,11 +39,15 @@ #include <grpc/support/log.h> #include <grpc/support/string_util.h> +#include "src/core/lib/channel/channel_args.h" #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/support/string.h" /* -- Fake transport security credentials. -- */ +#define GRPC_ARG_FAKE_SECURITY_EXPECTED_TARGETS \ + "grpc.fake_security.expected_targets" + static grpc_security_status fake_transport_security_create_security_connector( grpc_exec_ctx *exec_ctx, grpc_channel_credentials *c, grpc_call_credentials *call_creds, const char *target, @@ -88,6 +92,25 @@ grpc_server_credentials *grpc_fake_transport_security_server_credentials_create( return c; } +grpc_arg grpc_fake_transport_expected_targets_arg(char *expected_targets) { + grpc_arg arg; + arg.type = GRPC_ARG_STRING; + arg.key = GRPC_ARG_FAKE_SECURITY_EXPECTED_TARGETS; + arg.value.string = expected_targets; + return arg; +} + +const char *grpc_fake_transport_get_expected_targets( + const grpc_channel_args *args) { + const grpc_arg *expected_target_arg = + grpc_channel_args_find(args, GRPC_ARG_FAKE_SECURITY_EXPECTED_TARGETS); + if (expected_target_arg != NULL && + expected_target_arg->type == GRPC_ARG_STRING) { + return expected_target_arg->value.string; + } + return NULL; +} + /* -- Metadata-only test credentials. -- */ static void md_only_test_destruct(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/security/credentials/fake/fake_credentials.h b/src/core/lib/security/credentials/fake/fake_credentials.h index 0fe98417c6..a28b545a67 100644 --- a/src/core/lib/security/credentials/fake/fake_credentials.h +++ b/src/core/lib/security/credentials/fake/fake_credentials.h @@ -38,10 +38,17 @@ /* -- Fake transport security credentials. -- */ +/* Creates a fake transport security credentials object for testing. */ +grpc_channel_credentials *grpc_fake_transport_security_credentials_create(void); + +/* Creates a fake server transport security credentials object for testing. */ +grpc_server_credentials *grpc_fake_transport_security_server_credentials_create( + void); + /* Used to verify the target names given to the fake transport security * connector. * - * Its syntax by example: + * The syntax of \a expected_targets by example: * For LB channels: * "backend_target_1,backend_target_2,...;lb_target_1,lb_target_2,..." * For regular channels: @@ -50,15 +57,11 @@ * That is to say, LB channels have a heading list of LB targets separated from * the list of backend targets by a semicolon. For non-LB channels, only the * latter is present. */ -#define GRPC_ARG_FAKE_SECURITY_EXPECTED_TARGETS \ - "grpc.test_only.fake_security.expected_target" +grpc_arg grpc_fake_transport_expected_targets_arg(char *expected_targets); -/* Creates a fake transport security credentials object for testing. */ -grpc_channel_credentials *grpc_fake_transport_security_credentials_create(void); - -/* Creates a fake server transport security credentials object for testing. */ -grpc_server_credentials *grpc_fake_transport_security_server_credentials_create( - void); +/* Return the value associated with the expected targets channel arg or NULL */ +const char *grpc_fake_transport_get_expected_targets( + const grpc_channel_args *args); /* -- Metadata-only Test credentials. -- */ diff --git a/src/core/lib/security/transport/security_connector.c b/src/core/lib/security/transport/security_connector.c index dbe3263f92..b15196e677 100644 --- a/src/core/lib/security/transport/security_connector.c +++ b/src/core/lib/security/transport/security_connector.c @@ -423,12 +423,8 @@ grpc_channel_security_connector *grpc_fake_channel_security_connector_create( c->base.check_call_host = fake_channel_check_call_host; c->base.add_handshakers = fake_channel_add_handshakers; c->target = gpr_strdup(target); - const grpc_arg *expected_target_arg = - grpc_channel_args_find(args, GRPC_ARG_FAKE_SECURITY_EXPECTED_TARGETS); - if (expected_target_arg != NULL) { - GPR_ASSERT(expected_target_arg->type == GRPC_ARG_STRING); - c->expected_targets = gpr_strdup(expected_target_arg->value.string); - } + const char *expected_targets = grpc_fake_transport_get_expected_targets(args); + c->expected_targets = gpr_strdup(expected_targets); c->is_lb_channel = (grpc_lb_targets_info_find_in_args(args) != NULL); return &c->base; } diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index 6a8ae03a21..ef159a4ca4 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -1197,6 +1197,7 @@ static void finish_batch_step(grpc_exec_ctx *exec_ctx, batch_control *bctl) { static void continue_receiving_slices(grpc_exec_ctx *exec_ctx, batch_control *bctl) { + grpc_error *error; grpc_call *call = bctl->call; for (;;) { size_t remaining = call->receiving_stream->length - @@ -1208,11 +1209,22 @@ static void continue_receiving_slices(grpc_exec_ctx *exec_ctx, finish_batch_step(exec_ctx, bctl); return; } - if (grpc_byte_stream_next(exec_ctx, call->receiving_stream, - &call->receiving_slice, remaining, + if (grpc_byte_stream_next(exec_ctx, call->receiving_stream, remaining, &call->receiving_slice_ready)) { - grpc_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer, - call->receiving_slice); + error = grpc_byte_stream_pull(exec_ctx, call->receiving_stream, + &call->receiving_slice); + if (error == GRPC_ERROR_NONE) { + grpc_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer, + call->receiving_slice); + } else { + grpc_byte_stream_destroy(exec_ctx, call->receiving_stream); + call->receiving_stream = NULL; + grpc_byte_buffer_destroy(*call->receiving_buffer); + *call->receiving_buffer = NULL; + call->receiving_message = 0; + finish_batch_step(exec_ctx, bctl); + return; + } } else { return; } @@ -1223,12 +1235,24 @@ static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp, grpc_error *error) { batch_control *bctl = bctlp; grpc_call *call = bctl->call; + grpc_byte_stream *bs = call->receiving_stream; + bool release_error = false; if (error == GRPC_ERROR_NONE) { - grpc_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer, - call->receiving_slice); - continue_receiving_slices(exec_ctx, bctl); - } else { + grpc_slice slice; + error = grpc_byte_stream_pull(exec_ctx, bs, &slice); + if (error == GRPC_ERROR_NONE) { + grpc_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer, + slice); + continue_receiving_slices(exec_ctx, bctl); + } else { + /* Error returned by grpc_byte_stream_pull needs to be released manually + */ + release_error = true; + } + } + + if (error != GRPC_ERROR_NONE) { if (grpc_trace_operation_failures) { GRPC_LOG_IF_ERROR("receiving_slice_ready", GRPC_ERROR_REF(error)); } @@ -1236,7 +1260,11 @@ static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp, call->receiving_stream = NULL; grpc_byte_buffer_destroy(*call->receiving_buffer); *call->receiving_buffer = NULL; + call->receiving_message = 0; finish_batch_step(exec_ctx, bctl); + if (release_error) { + GRPC_ERROR_UNREF(error); + } } } diff --git a/src/core/lib/transport/byte_stream.c b/src/core/lib/transport/byte_stream.c index 4d4206189e..5800c70ef4 100644 --- a/src/core/lib/transport/byte_stream.c +++ b/src/core/lib/transport/byte_stream.c @@ -40,10 +40,15 @@ #include "src/core/lib/slice/slice_internal.h" int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx, - grpc_byte_stream *byte_stream, grpc_slice *slice, - size_t max_size_hint, grpc_closure *on_complete) { - return byte_stream->next(exec_ctx, byte_stream, slice, max_size_hint, - on_complete); + grpc_byte_stream *byte_stream, size_t max_size_hint, + grpc_closure *on_complete) { + return byte_stream->next(exec_ctx, byte_stream, max_size_hint, on_complete); +} + +grpc_error *grpc_byte_stream_pull(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + grpc_slice *slice) { + return byte_stream->pull(exec_ctx, byte_stream, slice); } void grpc_byte_stream_destroy(grpc_exec_ctx *exec_ctx, @@ -53,16 +58,24 @@ void grpc_byte_stream_destroy(grpc_exec_ctx *exec_ctx, /* slice_buffer_stream */ -static int slice_buffer_stream_next(grpc_exec_ctx *exec_ctx, - grpc_byte_stream *byte_stream, - grpc_slice *slice, size_t max_size_hint, - grpc_closure *on_complete) { +static bool slice_buffer_stream_next(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + size_t max_size_hint, + grpc_closure *on_complete) { + grpc_slice_buffer_stream *stream = (grpc_slice_buffer_stream *)byte_stream; + GPR_ASSERT(stream->cursor < stream->backing_buffer->count); + return true; +} + +static grpc_error *slice_buffer_stream_pull(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + grpc_slice *slice) { grpc_slice_buffer_stream *stream = (grpc_slice_buffer_stream *)byte_stream; GPR_ASSERT(stream->cursor < stream->backing_buffer->count); *slice = grpc_slice_ref_internal(stream->backing_buffer->slices[stream->cursor]); stream->cursor++; - return 1; + return GRPC_ERROR_NONE; } static void slice_buffer_stream_destroy(grpc_exec_ctx *exec_ctx, @@ -75,6 +88,7 @@ void grpc_slice_buffer_stream_init(grpc_slice_buffer_stream *stream, stream->base.length = (uint32_t)slice_buffer->length; stream->base.flags = flags; stream->base.next = slice_buffer_stream_next; + stream->base.pull = slice_buffer_stream_pull; stream->base.destroy = slice_buffer_stream_destroy; stream->backing_buffer = slice_buffer; stream->cursor = 0; diff --git a/src/core/lib/transport/byte_stream.h b/src/core/lib/transport/byte_stream.h index 1fdd5b4d77..381c65fb04 100644 --- a/src/core/lib/transport/byte_stream.h +++ b/src/core/lib/transport/byte_stream.h @@ -49,9 +49,10 @@ typedef struct grpc_byte_stream grpc_byte_stream; struct grpc_byte_stream { uint32_t length; uint32_t flags; - int (*next)(grpc_exec_ctx *exec_ctx, grpc_byte_stream *byte_stream, - grpc_slice *slice, size_t max_size_hint, - grpc_closure *on_complete); + bool (*next)(grpc_exec_ctx *exec_ctx, grpc_byte_stream *byte_stream, + size_t max_size_hint, grpc_closure *on_complete); + grpc_error *(*pull)(grpc_exec_ctx *exec_ctx, grpc_byte_stream *byte_stream, + grpc_slice *slice); void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_byte_stream *byte_stream); }; @@ -61,12 +62,20 @@ struct grpc_byte_stream { * * max_size_hint can be set as a hint as to the maximum number * of bytes that would be acceptable to read. + */ +int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, size_t max_size_hint, + grpc_closure *on_complete); + +/* returns the next slice in the byte stream when it is ready (indicated by + * either grpc_byte_stream_next returning 1 or on_complete passed to + * grpc_byte_stream_next is called). * * once a slice is returned into *slice, it is owned by the caller. */ -int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx, - grpc_byte_stream *byte_stream, grpc_slice *slice, - size_t max_size_hint, grpc_closure *on_complete); +grpc_error *grpc_byte_stream_pull(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + grpc_slice *slice); void grpc_byte_stream_destroy(grpc_exec_ctx *exec_ctx, grpc_byte_stream *byte_stream); diff --git a/src/core/lib/transport/static_metadata.c b/src/core/lib/transport/static_metadata.c index 599d4ad117..862cdaa8e0 100644 --- a/src/core/lib/transport/static_metadata.c +++ b/src/core/lib/transport/static_metadata.c @@ -97,23 +97,23 @@ static uint8_t g_bytes[] = { 101, 105, 102, 45, 110, 111, 110, 101, 45, 109, 97, 116, 99, 104, 105, 102, 45, 114, 97, 110, 103, 101, 105, 102, 45, 117, 110, 109, 111, 100, 105, 102, 105, 101, 100, 45, 115, 105, 110, 99, 101, 108, 97, 115, 116, - 45, 109, 111, 100, 105, 102, 105, 101, 100, 108, 105, 110, 107, 108, 111, - 99, 97, 116, 105, 111, 110, 109, 97, 120, 45, 102, 111, 114, 119, 97, - 114, 100, 115, 112, 114, 111, 120, 121, 45, 97, 117, 116, 104, 101, 110, - 116, 105, 99, 97, 116, 101, 112, 114, 111, 120, 121, 45, 97, 117, 116, - 104, 111, 114, 105, 122, 97, 116, 105, 111, 110, 114, 97, 110, 103, 101, - 114, 101, 102, 101, 114, 101, 114, 114, 101, 102, 114, 101, 115, 104, 114, - 101, 116, 114, 121, 45, 97, 102, 116, 101, 114, 115, 101, 114, 118, 101, - 114, 115, 101, 116, 45, 99, 111, 111, 107, 105, 101, 115, 116, 114, 105, - 99, 116, 45, 116, 114, 97, 110, 115, 112, 111, 114, 116, 45, 115, 101, - 99, 117, 114, 105, 116, 121, 116, 114, 97, 110, 115, 102, 101, 114, 45, - 101, 110, 99, 111, 100, 105, 110, 103, 118, 97, 114, 121, 118, 105, 97, - 119, 119, 119, 45, 97, 117, 116, 104, 101, 110, 116, 105, 99, 97, 116, - 101, 105, 100, 101, 110, 116, 105, 116, 121, 44, 100, 101, 102, 108, 97, - 116, 101, 105, 100, 101, 110, 116, 105, 116, 121, 44, 103, 122, 105, 112, - 100, 101, 102, 108, 97, 116, 101, 44, 103, 122, 105, 112, 105, 100, 101, - 110, 116, 105, 116, 121, 44, 100, 101, 102, 108, 97, 116, 101, 44, 103, - 122, 105, 112}; + 45, 109, 111, 100, 105, 102, 105, 101, 100, 108, 98, 45, 99, 111, 115, + 116, 45, 98, 105, 110, 108, 105, 110, 107, 108, 111, 99, 97, 116, 105, + 111, 110, 109, 97, 120, 45, 102, 111, 114, 119, 97, 114, 100, 115, 112, + 114, 111, 120, 121, 45, 97, 117, 116, 104, 101, 110, 116, 105, 99, 97, + 116, 101, 112, 114, 111, 120, 121, 45, 97, 117, 116, 104, 111, 114, 105, + 122, 97, 116, 105, 111, 110, 114, 97, 110, 103, 101, 114, 101, 102, 101, + 114, 101, 114, 114, 101, 102, 114, 101, 115, 104, 114, 101, 116, 114, 121, + 45, 97, 102, 116, 101, 114, 115, 101, 114, 118, 101, 114, 115, 101, 116, + 45, 99, 111, 111, 107, 105, 101, 115, 116, 114, 105, 99, 116, 45, 116, + 114, 97, 110, 115, 112, 111, 114, 116, 45, 115, 101, 99, 117, 114, 105, + 116, 121, 116, 114, 97, 110, 115, 102, 101, 114, 45, 101, 110, 99, 111, + 100, 105, 110, 103, 118, 97, 114, 121, 118, 105, 97, 119, 119, 119, 45, + 97, 117, 116, 104, 101, 110, 116, 105, 99, 97, 116, 101, 105, 100, 101, + 110, 116, 105, 116, 121, 44, 100, 101, 102, 108, 97, 116, 101, 105, 100, + 101, 110, 116, 105, 116, 121, 44, 103, 122, 105, 112, 100, 101, 102, 108, + 97, 116, 101, 44, 103, 122, 105, 112, 105, 100, 101, 110, 116, 105, 116, + 121, 44, 100, 101, 102, 108, 97, 116, 101, 44, 103, 122, 105, 112}; static void static_ref(void *unused) {} static void static_unref(grpc_exec_ctx *exec_ctx, void *unused) {} @@ -223,6 +223,7 @@ grpc_slice_refcount grpc_static_metadata_refcounts[GRPC_STATIC_MDSTR_COUNT] = { {&grpc_static_metadata_vtable, &static_sub_refcnt}, {&grpc_static_metadata_vtable, &static_sub_refcnt}, {&grpc_static_metadata_vtable, &static_sub_refcnt}, + {&grpc_static_metadata_vtable, &static_sub_refcnt}, }; const grpc_slice grpc_static_slice_table[GRPC_STATIC_MDSTR_COUNT] = { @@ -383,64 +384,67 @@ const grpc_slice grpc_static_slice_table[GRPC_STATIC_MDSTR_COUNT] = { {.refcount = &grpc_static_metadata_refcounts[77], .data.refcounted = {g_bytes + 791, 13}}, {.refcount = &grpc_static_metadata_refcounts[78], - .data.refcounted = {g_bytes + 804, 4}}, + .data.refcounted = {g_bytes + 804, 11}}, {.refcount = &grpc_static_metadata_refcounts[79], - .data.refcounted = {g_bytes + 808, 8}}, + .data.refcounted = {g_bytes + 815, 4}}, {.refcount = &grpc_static_metadata_refcounts[80], - .data.refcounted = {g_bytes + 816, 12}}, + .data.refcounted = {g_bytes + 819, 8}}, {.refcount = &grpc_static_metadata_refcounts[81], - .data.refcounted = {g_bytes + 828, 18}}, + .data.refcounted = {g_bytes + 827, 12}}, {.refcount = &grpc_static_metadata_refcounts[82], - .data.refcounted = {g_bytes + 846, 19}}, + .data.refcounted = {g_bytes + 839, 18}}, {.refcount = &grpc_static_metadata_refcounts[83], - .data.refcounted = {g_bytes + 865, 5}}, + .data.refcounted = {g_bytes + 857, 19}}, {.refcount = &grpc_static_metadata_refcounts[84], - .data.refcounted = {g_bytes + 870, 7}}, + .data.refcounted = {g_bytes + 876, 5}}, {.refcount = &grpc_static_metadata_refcounts[85], - .data.refcounted = {g_bytes + 877, 7}}, + .data.refcounted = {g_bytes + 881, 7}}, {.refcount = &grpc_static_metadata_refcounts[86], - .data.refcounted = {g_bytes + 884, 11}}, + .data.refcounted = {g_bytes + 888, 7}}, {.refcount = &grpc_static_metadata_refcounts[87], - .data.refcounted = {g_bytes + 895, 6}}, + .data.refcounted = {g_bytes + 895, 11}}, {.refcount = &grpc_static_metadata_refcounts[88], - .data.refcounted = {g_bytes + 901, 10}}, + .data.refcounted = {g_bytes + 906, 6}}, {.refcount = &grpc_static_metadata_refcounts[89], - .data.refcounted = {g_bytes + 911, 25}}, + .data.refcounted = {g_bytes + 912, 10}}, {.refcount = &grpc_static_metadata_refcounts[90], - .data.refcounted = {g_bytes + 936, 17}}, + .data.refcounted = {g_bytes + 922, 25}}, {.refcount = &grpc_static_metadata_refcounts[91], - .data.refcounted = {g_bytes + 953, 4}}, + .data.refcounted = {g_bytes + 947, 17}}, {.refcount = &grpc_static_metadata_refcounts[92], - .data.refcounted = {g_bytes + 957, 3}}, + .data.refcounted = {g_bytes + 964, 4}}, {.refcount = &grpc_static_metadata_refcounts[93], - .data.refcounted = {g_bytes + 960, 16}}, + .data.refcounted = {g_bytes + 968, 3}}, {.refcount = &grpc_static_metadata_refcounts[94], - .data.refcounted = {g_bytes + 976, 16}}, + .data.refcounted = {g_bytes + 971, 16}}, {.refcount = &grpc_static_metadata_refcounts[95], - .data.refcounted = {g_bytes + 992, 13}}, + .data.refcounted = {g_bytes + 987, 16}}, {.refcount = &grpc_static_metadata_refcounts[96], - .data.refcounted = {g_bytes + 1005, 12}}, + .data.refcounted = {g_bytes + 1003, 13}}, {.refcount = &grpc_static_metadata_refcounts[97], - .data.refcounted = {g_bytes + 1017, 21}}, + .data.refcounted = {g_bytes + 1016, 12}}, + {.refcount = &grpc_static_metadata_refcounts[98], + .data.refcounted = {g_bytes + 1028, 21}}, }; uintptr_t grpc_static_mdelem_user_data[GRPC_STATIC_MDELEM_COUNT] = { - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2, 4, 4, 6, 6, 8, 8}; + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2, 4, 4, 6, 6, 8, 8}; static const int8_t elems_r[] = { - 10, 8, -3, 0, 9, 21, -76, 22, 0, 10, -7, 0, 0, 0, 14, 0, - 13, 12, 11, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, -50, -51, 15, -53, -54, -55, -56, -56, -57, -58, 0, 37, 36, 35, 34, - 33, 32, 31, 30, 29, 28, 27, 26, 25, 24, 23, 22, 21, 20, 19, 18, - 17, 16, 15, 14, 13, 12, 11, 10, 13, 12, 11, 10, 9, 8, 7, 0}; + 10, 8, -3, 0, 9, 21, -77, 22, 0, 10, -7, 0, 0, 0, + 14, 0, 13, 12, 11, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, -50, -51, 16, -53, -54, -55, -56, + -56, -57, -58, -59, 0, 37, 36, 35, 34, 33, 32, 31, 30, 29, + 28, 27, 26, 25, 24, 23, 22, 21, 20, 19, 18, 17, 16, 15, + 14, 13, 12, 11, 10, 13, 12, 11, 10, 9, 8, 7, 0}; static uint32_t elems_phash(uint32_t i) { i -= 42; - uint32_t x = i % 96; - uint32_t y = i / 96; + uint32_t x = i % 97; + uint32_t y = i / 97; uint32_t h = x; if (y < GPR_ARRAY_SIZE(elems_r)) { uint32_t delta = (uint32_t)elems_r[y]; @@ -450,29 +454,30 @@ static uint32_t elems_phash(uint32_t i) { } static const uint16_t elem_keys[] = { - 1009, 1010, 1011, 240, 241, 242, 243, 244, 138, 139, 42, 43, - 429, 430, 431, 911, 912, 913, 712, 713, 1392, 522, 714, 1588, - 1686, 1784, 4822, 4920, 4951, 5116, 5214, 5312, 5410, 1405, 5508, 5606, - 5704, 5802, 5900, 5998, 6096, 6194, 6292, 6390, 6488, 6586, 6684, 6782, - 6880, 6978, 7076, 7174, 7272, 7370, 7468, 7566, 7664, 7762, 7860, 7958, - 8056, 8154, 8252, 8350, 8448, 1074, 1075, 1076, 1077, 8546, 8644, 8742, - 8840, 8938, 9036, 9134, 0, 314, 0, 0, 0, 0, 0, 0, + 1019, 1020, 1021, 242, 243, 244, 245, 246, 139, 140, 42, 43, + 433, 434, 435, 920, 921, 922, 719, 720, 1406, 527, 721, 1604, + 1703, 1802, 4871, 4970, 5001, 5168, 5267, 5366, 5465, 1419, 5564, 5663, + 5762, 5861, 5960, 6059, 6158, 6257, 6356, 6455, 6554, 6653, 6752, 6851, + 6950, 7049, 7148, 7247, 7346, 7445, 7544, 7643, 7742, 7841, 7940, 8039, + 8138, 8237, 8336, 8435, 8534, 8633, 1085, 1086, 1087, 1088, 8732, 8831, + 8930, 9029, 9128, 9227, 9326, 0, 317, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 133, 233, 234, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 132, 231, 232, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; + 0}; static const uint8_t elem_idxs[] = { - 73, 76, 74, 19, 20, 21, 22, 23, 15, 16, 17, 18, 11, 12, 13, - 3, 4, 5, 0, 1, 41, 6, 2, 69, 48, 55, 24, 25, 26, 27, + 74, 77, 75, 19, 20, 21, 22, 23, 15, 16, 17, 18, 11, 12, 13, + 3, 4, 5, 0, 1, 41, 6, 2, 70, 48, 55, 24, 25, 26, 27, 28, 29, 30, 7, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 42, 43, 44, 45, 46, 47, 49, 50, 51, 52, 53, 54, 56, 57, 58, 59, - 60, 61, 62, 63, 64, 75, 77, 78, 79, 65, 66, 67, 68, 70, 71, - 72, 255, 14, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, - 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 8, 9, 10}; + 60, 61, 62, 63, 64, 65, 76, 78, 79, 80, 66, 67, 68, 69, 71, + 72, 73, 255, 14, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, + 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 8, 9, 10}; grpc_mdelem grpc_static_mdelem_for_static_strings(int a, int b) { if (a == -1 || b == -1) return GRPC_MDNULL; - uint32_t k = (uint32_t)(a * 98 + b); + uint32_t k = (uint32_t)(a * 99 + b); uint32_t h = elems_phash(k); return h < GPR_ARRAY_SIZE(elem_keys) && elem_keys[h] == k ? GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[elem_idxs[h]], @@ -706,71 +711,75 @@ grpc_mdelem_data grpc_static_mdelem_table[GRPC_STATIC_MDELEM_COUNT] = { {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[78], - .data.refcounted = {g_bytes + 804, 4}}, + .data.refcounted = {g_bytes + 804, 11}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[79], - .data.refcounted = {g_bytes + 808, 8}}, + .data.refcounted = {g_bytes + 815, 4}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[80], - .data.refcounted = {g_bytes + 816, 12}}, + .data.refcounted = {g_bytes + 819, 8}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[81], - .data.refcounted = {g_bytes + 828, 18}}, + .data.refcounted = {g_bytes + 827, 12}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[82], - .data.refcounted = {g_bytes + 846, 19}}, + .data.refcounted = {g_bytes + 839, 18}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[83], - .data.refcounted = {g_bytes + 865, 5}}, + .data.refcounted = {g_bytes + 857, 19}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[84], - .data.refcounted = {g_bytes + 870, 7}}, + .data.refcounted = {g_bytes + 876, 5}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[85], - .data.refcounted = {g_bytes + 877, 7}}, + .data.refcounted = {g_bytes + 881, 7}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[86], - .data.refcounted = {g_bytes + 884, 11}}, + .data.refcounted = {g_bytes + 888, 7}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[87], - .data.refcounted = {g_bytes + 895, 6}}, + .data.refcounted = {g_bytes + 895, 11}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[88], - .data.refcounted = {g_bytes + 901, 10}}, + .data.refcounted = {g_bytes + 906, 6}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[89], - .data.refcounted = {g_bytes + 911, 25}}, + .data.refcounted = {g_bytes + 912, 10}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[90], - .data.refcounted = {g_bytes + 936, 17}}, + .data.refcounted = {g_bytes + 922, 25}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, - {{.refcount = &grpc_static_metadata_refcounts[16], - .data.refcounted = {g_bytes + 200, 10}}, + {{.refcount = &grpc_static_metadata_refcounts[91], + .data.refcounted = {g_bytes + 947, 17}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, - {{.refcount = &grpc_static_metadata_refcounts[91], - .data.refcounted = {g_bytes + 953, 4}}, + {{.refcount = &grpc_static_metadata_refcounts[16], + .data.refcounted = {g_bytes + 200, 10}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[92], - .data.refcounted = {g_bytes + 957, 3}}, + .data.refcounted = {g_bytes + 964, 4}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[93], - .data.refcounted = {g_bytes + 960, 16}}, + .data.refcounted = {g_bytes + 968, 3}}, + {.refcount = &grpc_static_metadata_refcounts[20], + .data.refcounted = {g_bytes + 234, 0}}}, + {{.refcount = &grpc_static_metadata_refcounts[94], + .data.refcounted = {g_bytes + 971, 16}}, {.refcount = &grpc_static_metadata_refcounts[20], .data.refcounted = {g_bytes + 234, 0}}}, {{.refcount = &grpc_static_metadata_refcounts[10], @@ -783,24 +792,24 @@ grpc_mdelem_data grpc_static_mdelem_table[GRPC_STATIC_MDELEM_COUNT] = { .data.refcounted = {g_bytes + 377, 7}}}, {{.refcount = &grpc_static_metadata_refcounts[10], .data.refcounted = {g_bytes + 90, 20}}, - {.refcount = &grpc_static_metadata_refcounts[94], - .data.refcounted = {g_bytes + 976, 16}}}, + {.refcount = &grpc_static_metadata_refcounts[95], + .data.refcounted = {g_bytes + 987, 16}}}, {{.refcount = &grpc_static_metadata_refcounts[10], .data.refcounted = {g_bytes + 90, 20}}, {.refcount = &grpc_static_metadata_refcounts[30], .data.refcounted = {g_bytes + 373, 4}}}, {{.refcount = &grpc_static_metadata_refcounts[10], .data.refcounted = {g_bytes + 90, 20}}, - {.refcount = &grpc_static_metadata_refcounts[95], - .data.refcounted = {g_bytes + 992, 13}}}, - {{.refcount = &grpc_static_metadata_refcounts[10], - .data.refcounted = {g_bytes + 90, 20}}, {.refcount = &grpc_static_metadata_refcounts[96], - .data.refcounted = {g_bytes + 1005, 12}}}, + .data.refcounted = {g_bytes + 1003, 13}}}, {{.refcount = &grpc_static_metadata_refcounts[10], .data.refcounted = {g_bytes + 90, 20}}, {.refcount = &grpc_static_metadata_refcounts[97], - .data.refcounted = {g_bytes + 1017, 21}}}, + .data.refcounted = {g_bytes + 1016, 12}}}, + {{.refcount = &grpc_static_metadata_refcounts[10], + .data.refcounted = {g_bytes + 90, 20}}, + {.refcount = &grpc_static_metadata_refcounts[98], + .data.refcounted = {g_bytes + 1028, 21}}}, }; -const uint8_t grpc_static_accept_encoding_metadata[8] = {0, 73, 74, 75, - 76, 77, 78, 79}; +const uint8_t grpc_static_accept_encoding_metadata[8] = {0, 74, 75, 76, + 77, 78, 79, 80}; diff --git a/src/core/lib/transport/static_metadata.h b/src/core/lib/transport/static_metadata.h index dcbe418edc..84fb316fd6 100644 --- a/src/core/lib/transport/static_metadata.h +++ b/src/core/lib/transport/static_metadata.h @@ -44,7 +44,7 @@ #include "src/core/lib/transport/metadata.h" -#define GRPC_STATIC_MDSTR_COUNT 98 +#define GRPC_STATIC_MDSTR_COUNT 99 extern const grpc_slice grpc_static_slice_table[GRPC_STATIC_MDSTR_COUNT]; /* ":path" */ #define GRPC_MDSTR_PATH (grpc_static_slice_table[0]) @@ -205,47 +205,49 @@ extern const grpc_slice grpc_static_slice_table[GRPC_STATIC_MDSTR_COUNT]; #define GRPC_MDSTR_IF_UNMODIFIED_SINCE (grpc_static_slice_table[76]) /* "last-modified" */ #define GRPC_MDSTR_LAST_MODIFIED (grpc_static_slice_table[77]) +/* "lb-cost-bin" */ +#define GRPC_MDSTR_LB_COST_BIN (grpc_static_slice_table[78]) /* "link" */ -#define GRPC_MDSTR_LINK (grpc_static_slice_table[78]) +#define GRPC_MDSTR_LINK (grpc_static_slice_table[79]) /* "location" */ -#define GRPC_MDSTR_LOCATION (grpc_static_slice_table[79]) +#define GRPC_MDSTR_LOCATION (grpc_static_slice_table[80]) /* "max-forwards" */ -#define GRPC_MDSTR_MAX_FORWARDS (grpc_static_slice_table[80]) +#define GRPC_MDSTR_MAX_FORWARDS (grpc_static_slice_table[81]) /* "proxy-authenticate" */ -#define GRPC_MDSTR_PROXY_AUTHENTICATE (grpc_static_slice_table[81]) +#define GRPC_MDSTR_PROXY_AUTHENTICATE (grpc_static_slice_table[82]) /* "proxy-authorization" */ -#define GRPC_MDSTR_PROXY_AUTHORIZATION (grpc_static_slice_table[82]) +#define GRPC_MDSTR_PROXY_AUTHORIZATION (grpc_static_slice_table[83]) /* "range" */ -#define GRPC_MDSTR_RANGE (grpc_static_slice_table[83]) +#define GRPC_MDSTR_RANGE (grpc_static_slice_table[84]) /* "referer" */ -#define GRPC_MDSTR_REFERER (grpc_static_slice_table[84]) +#define GRPC_MDSTR_REFERER (grpc_static_slice_table[85]) /* "refresh" */ -#define GRPC_MDSTR_REFRESH (grpc_static_slice_table[85]) +#define GRPC_MDSTR_REFRESH (grpc_static_slice_table[86]) /* "retry-after" */ -#define GRPC_MDSTR_RETRY_AFTER (grpc_static_slice_table[86]) +#define GRPC_MDSTR_RETRY_AFTER (grpc_static_slice_table[87]) /* "server" */ -#define GRPC_MDSTR_SERVER (grpc_static_slice_table[87]) +#define GRPC_MDSTR_SERVER (grpc_static_slice_table[88]) /* "set-cookie" */ -#define GRPC_MDSTR_SET_COOKIE (grpc_static_slice_table[88]) +#define GRPC_MDSTR_SET_COOKIE (grpc_static_slice_table[89]) /* "strict-transport-security" */ -#define GRPC_MDSTR_STRICT_TRANSPORT_SECURITY (grpc_static_slice_table[89]) +#define GRPC_MDSTR_STRICT_TRANSPORT_SECURITY (grpc_static_slice_table[90]) /* "transfer-encoding" */ -#define GRPC_MDSTR_TRANSFER_ENCODING (grpc_static_slice_table[90]) +#define GRPC_MDSTR_TRANSFER_ENCODING (grpc_static_slice_table[91]) /* "vary" */ -#define GRPC_MDSTR_VARY (grpc_static_slice_table[91]) +#define GRPC_MDSTR_VARY (grpc_static_slice_table[92]) /* "via" */ -#define GRPC_MDSTR_VIA (grpc_static_slice_table[92]) +#define GRPC_MDSTR_VIA (grpc_static_slice_table[93]) /* "www-authenticate" */ -#define GRPC_MDSTR_WWW_AUTHENTICATE (grpc_static_slice_table[93]) +#define GRPC_MDSTR_WWW_AUTHENTICATE (grpc_static_slice_table[94]) /* "identity,deflate" */ -#define GRPC_MDSTR_IDENTITY_COMMA_DEFLATE (grpc_static_slice_table[94]) +#define GRPC_MDSTR_IDENTITY_COMMA_DEFLATE (grpc_static_slice_table[95]) /* "identity,gzip" */ -#define GRPC_MDSTR_IDENTITY_COMMA_GZIP (grpc_static_slice_table[95]) +#define GRPC_MDSTR_IDENTITY_COMMA_GZIP (grpc_static_slice_table[96]) /* "deflate,gzip" */ -#define GRPC_MDSTR_DEFLATE_COMMA_GZIP (grpc_static_slice_table[96]) +#define GRPC_MDSTR_DEFLATE_COMMA_GZIP (grpc_static_slice_table[97]) /* "identity,deflate,gzip" */ #define GRPC_MDSTR_IDENTITY_COMMA_DEFLATE_COMMA_GZIP \ - (grpc_static_slice_table[97]) + (grpc_static_slice_table[98]) extern const grpc_slice_refcount_vtable grpc_static_metadata_vtable; extern grpc_slice_refcount @@ -257,7 +259,7 @@ extern grpc_slice_refcount #define GRPC_STATIC_METADATA_INDEX(static_slice) \ ((int)((static_slice).refcount - grpc_static_metadata_refcounts)) -#define GRPC_STATIC_MDELEM_COUNT 80 +#define GRPC_STATIC_MDELEM_COUNT 81 extern grpc_mdelem_data grpc_static_mdelem_table[GRPC_STATIC_MDELEM_COUNT]; extern uintptr_t grpc_static_mdelem_user_data[GRPC_STATIC_MDELEM_COUNT]; /* "grpc-status": "0" */ @@ -428,78 +430,81 @@ extern uintptr_t grpc_static_mdelem_user_data[GRPC_STATIC_MDELEM_COUNT]; /* "lb-token": "" */ #define GRPC_MDELEM_LB_TOKEN_EMPTY \ (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[55], GRPC_MDELEM_STORAGE_STATIC)) +/* "lb-cost-bin": "" */ +#define GRPC_MDELEM_LB_COST_BIN_EMPTY \ + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[56], GRPC_MDELEM_STORAGE_STATIC)) /* "link": "" */ #define GRPC_MDELEM_LINK_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[56], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[57], GRPC_MDELEM_STORAGE_STATIC)) /* "location": "" */ #define GRPC_MDELEM_LOCATION_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[57], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[58], GRPC_MDELEM_STORAGE_STATIC)) /* "max-forwards": "" */ #define GRPC_MDELEM_MAX_FORWARDS_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[58], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[59], GRPC_MDELEM_STORAGE_STATIC)) /* "proxy-authenticate": "" */ #define GRPC_MDELEM_PROXY_AUTHENTICATE_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[59], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[60], GRPC_MDELEM_STORAGE_STATIC)) /* "proxy-authorization": "" */ #define GRPC_MDELEM_PROXY_AUTHORIZATION_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[60], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[61], GRPC_MDELEM_STORAGE_STATIC)) /* "range": "" */ #define GRPC_MDELEM_RANGE_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[61], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[62], GRPC_MDELEM_STORAGE_STATIC)) /* "referer": "" */ #define GRPC_MDELEM_REFERER_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[62], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[63], GRPC_MDELEM_STORAGE_STATIC)) /* "refresh": "" */ #define GRPC_MDELEM_REFRESH_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[63], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[64], GRPC_MDELEM_STORAGE_STATIC)) /* "retry-after": "" */ #define GRPC_MDELEM_RETRY_AFTER_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[64], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[65], GRPC_MDELEM_STORAGE_STATIC)) /* "server": "" */ #define GRPC_MDELEM_SERVER_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[65], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[66], GRPC_MDELEM_STORAGE_STATIC)) /* "set-cookie": "" */ #define GRPC_MDELEM_SET_COOKIE_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[66], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[67], GRPC_MDELEM_STORAGE_STATIC)) /* "strict-transport-security": "" */ #define GRPC_MDELEM_STRICT_TRANSPORT_SECURITY_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[67], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[68], GRPC_MDELEM_STORAGE_STATIC)) /* "transfer-encoding": "" */ #define GRPC_MDELEM_TRANSFER_ENCODING_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[68], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[69], GRPC_MDELEM_STORAGE_STATIC)) /* "user-agent": "" */ #define GRPC_MDELEM_USER_AGENT_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[69], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[70], GRPC_MDELEM_STORAGE_STATIC)) /* "vary": "" */ #define GRPC_MDELEM_VARY_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[70], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[71], GRPC_MDELEM_STORAGE_STATIC)) /* "via": "" */ #define GRPC_MDELEM_VIA_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[71], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[72], GRPC_MDELEM_STORAGE_STATIC)) /* "www-authenticate": "" */ #define GRPC_MDELEM_WWW_AUTHENTICATE_EMPTY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[72], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[73], GRPC_MDELEM_STORAGE_STATIC)) /* "grpc-accept-encoding": "identity" */ #define GRPC_MDELEM_GRPC_ACCEPT_ENCODING_IDENTITY \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[73], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[74], GRPC_MDELEM_STORAGE_STATIC)) /* "grpc-accept-encoding": "deflate" */ #define GRPC_MDELEM_GRPC_ACCEPT_ENCODING_DEFLATE \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[74], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[75], GRPC_MDELEM_STORAGE_STATIC)) /* "grpc-accept-encoding": "identity,deflate" */ #define GRPC_MDELEM_GRPC_ACCEPT_ENCODING_IDENTITY_COMMA_DEFLATE \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[75], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[76], GRPC_MDELEM_STORAGE_STATIC)) /* "grpc-accept-encoding": "gzip" */ #define GRPC_MDELEM_GRPC_ACCEPT_ENCODING_GZIP \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[76], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[77], GRPC_MDELEM_STORAGE_STATIC)) /* "grpc-accept-encoding": "identity,gzip" */ #define GRPC_MDELEM_GRPC_ACCEPT_ENCODING_IDENTITY_COMMA_GZIP \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[77], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[78], GRPC_MDELEM_STORAGE_STATIC)) /* "grpc-accept-encoding": "deflate,gzip" */ #define GRPC_MDELEM_GRPC_ACCEPT_ENCODING_DEFLATE_COMMA_GZIP \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[78], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[79], GRPC_MDELEM_STORAGE_STATIC)) /* "grpc-accept-encoding": "identity,deflate,gzip" */ #define GRPC_MDELEM_GRPC_ACCEPT_ENCODING_IDENTITY_COMMA_DEFLATE_COMMA_GZIP \ - (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[79], GRPC_MDELEM_STORAGE_STATIC)) + (GRPC_MAKE_MDELEM(&grpc_static_mdelem_table[80], GRPC_MDELEM_STORAGE_STATIC)) grpc_mdelem grpc_static_mdelem_for_static_strings(int a, int b); typedef enum { diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc index cc50786b84..3c42afd2b1 100644 --- a/src/cpp/server/server_cc.cc +++ b/src/cpp/server/server_cc.cc @@ -503,7 +503,7 @@ int Server::AddListeningPort(const grpc::string& addr, ServerCredentials* creds) { GPR_ASSERT(!started_); int port = creds->AddPortToServer(addr, server_); - global_callbacks_->AddPort(this, port); + global_callbacks_->AddPort(this, addr, creds, port); return port; } diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 454cd886b0..923556413e 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -230,17 +230,9 @@ const struct census_context* ServerContext::census_context() const { void ServerContext::SetLoadReportingCosts( const std::vector<grpc::string>& cost_data) { if (call_ == nullptr) return; - grpc_load_reporting_cost_context* cost_ctx = - static_cast<grpc_load_reporting_cost_context*>( - gpr_malloc(sizeof(*cost_ctx))); - cost_ctx->values_count = cost_data.size(); - cost_ctx->values = static_cast<grpc_slice*>( - gpr_malloc(sizeof(*cost_ctx->values) * cost_ctx->values_count)); - for (size_t i = 0; i < cost_ctx->values_count; ++i) { - cost_ctx->values[i] = - grpc_slice_from_copied_buffer(cost_data[i].data(), cost_data[i].size()); + for (const auto& cost_datum : cost_data) { + AddTrailingMetadata(GRPC_LB_COST_MD_KEY, cost_datum); } - grpc_call_set_load_reporting_cost_context(call_, cost_ctx); } } // namespace grpc diff --git a/src/csharp/Grpc.Core.Tests/AppDomainUnloadTest.cs b/src/csharp/Grpc.Core.Tests/AppDomainUnloadTest.cs index d7ebdb4201..7858e77b27 100644 --- a/src/csharp/Grpc.Core.Tests/AppDomainUnloadTest.cs +++ b/src/csharp/Grpc.Core.Tests/AppDomainUnloadTest.cs @@ -72,10 +72,6 @@ namespace Grpc.Core.Tests public AppDomainTestClass() { var helper = new MockServiceHelper(Host); - var server = helper.GetServer(); - server.Start(); - var channel = helper.GetChannel(); - var readyToShutdown = new TaskCompletionSource<object>(); helper.DuplexStreamingHandler = new DuplexStreamingServerMethod<string, string>(async (requestStream, responseStream, context) => { @@ -83,6 +79,10 @@ namespace Grpc.Core.Tests await requestStream.ToListAsync(); }); + var server = helper.GetServer(); + server.Start(); + var channel = helper.GetChannel(); + var call = Calls.AsyncDuplexStreamingCall(helper.CreateDuplexStreamingCall()); readyToShutdown.Task.Wait(); // make sure handler is running } diff --git a/src/csharp/Grpc.Core.Tests/UserAgentStringTest.cs b/src/csharp/Grpc.Core.Tests/UserAgentStringTest.cs index cc830086a6..74b4997f69 100644 --- a/src/csharp/Grpc.Core.Tests/UserAgentStringTest.cs +++ b/src/csharp/Grpc.Core.Tests/UserAgentStringTest.cs @@ -63,10 +63,6 @@ namespace Grpc.Core.Tests public void DefaultUserAgentString() { helper = new MockServiceHelper(Host); - server = helper.GetServer(); - server.Start(); - channel = helper.GetChannel(); - helper.UnaryHandler = new UnaryServerMethod<string, string>((request, context) => { var userAgentString = context.RequestHeaders.First(m => (m.Key == "user-agent")).Value; @@ -75,6 +71,11 @@ namespace Grpc.Core.Tests Assert.IsTrue(parts[1].StartsWith("grpc-c/")); return Task.FromResult("PASS"); }); + + server = helper.GetServer(); + server.Start(); + channel = helper.GetChannel(); + Assert.AreEqual("PASS", Calls.BlockingUnaryCall(helper.CreateUnaryCall(), "")); } @@ -83,11 +84,6 @@ namespace Grpc.Core.Tests { helper = new MockServiceHelper(Host, channelOptions: new[] { new ChannelOption(ChannelOptions.PrimaryUserAgentString, "XYZ") }); - server = helper.GetServer(); - server.Start(); - channel = helper.GetChannel(); - - channel = helper.GetChannel(); helper.UnaryHandler = new UnaryServerMethod<string, string>((request, context) => { var userAgentString = context.RequestHeaders.First(m => (m.Key == "user-agent")).Value; @@ -95,6 +91,11 @@ namespace Grpc.Core.Tests Assert.AreEqual("XYZ", parts[0]); return Task.FromResult("PASS"); }); + + server = helper.GetServer(); + server.Start(); + channel = helper.GetChannel(); + Assert.AreEqual("PASS", Calls.BlockingUnaryCall(helper.CreateUnaryCall(), "")); } } diff --git a/src/objective-c/BoringSSL.podspec b/src/objective-c/BoringSSL.podspec index 7880aad10b..651bd4977d 100644 --- a/src/objective-c/BoringSSL.podspec +++ b/src/objective-c/BoringSSL.podspec @@ -31,7 +31,7 @@ Pod::Spec.new do |s| s.name = 'BoringSSL' - version = '8.1' + version = '8.2' s.version = version s.summary = 'BoringSSL is a fork of OpenSSL that is designed to meet Google’s needs.' # Adapted from the homepage: @@ -113,7 +113,6 @@ Pod::Spec.new do |s| s.subspec 'Interface' do |ss| ss.header_mappings_dir = 'include/openssl' ss.source_files = 'include/openssl/*.h' - ss.exclude_files = 'include/openssl/arm_arch.h' end s.subspec 'Implementation' do |ss| ss.header_mappings_dir = '.' @@ -150,6 +149,11 @@ Pod::Spec.new do |s| #include "ssl.h" #include "crypto.h" #include "aes.h" + /* The following macros are defined by base.h. The latter is the first file included by the + other headers. */ + #if defined(OPENSSL_ARM) || defined(OPENSSL_AARCH64) + # include "arm_arch.h" + #endif #include "asn1.h" #include "asn1_mac.h" #include "asn1t.h" diff --git a/src/objective-c/tests/InteropTestsLocalCleartext.m b/src/objective-c/tests/InteropTestsLocalCleartext.m index 94cdf85965..cdcdd8a88c 100644 --- a/src/objective-c/tests/InteropTestsLocalCleartext.m +++ b/src/objective-c/tests/InteropTestsLocalCleartext.m @@ -36,7 +36,11 @@ #import "InteropTests.h" -static NSString * const kLocalCleartextHost = @"localhost:5050"; +// The server address is derived from preprocessor macro, which is +// in turn derived from environment variable of the same name. +#define NSStringize_helper(x) #x +#define NSStringize(x) @NSStringize_helper(x) +static NSString * const kLocalCleartextHost = NSStringize(HOST_PORT_LOCAL); // The Protocol Buffers encoding overhead of local interop server. Acquired // by experiment. Adjust this when server's proto file changes. diff --git a/src/objective-c/tests/InteropTestsLocalSSL.m b/src/objective-c/tests/InteropTestsLocalSSL.m index 3c78b65ede..45e62d29b9 100644 --- a/src/objective-c/tests/InteropTestsLocalSSL.m +++ b/src/objective-c/tests/InteropTestsLocalSSL.m @@ -35,8 +35,11 @@ #import <GRPCClient/internal_testing/GRPCCall+InternalTests.h> #import "InteropTests.h" - -static NSString * const kLocalSSLHost = @"localhost:5051"; +// The server address is derived from preprocessor macro, which is +// in turn derived from environment variable of the same name. +#define NSStringize_helper(x) #x +#define NSStringize(x) @NSStringize_helper(x) +static NSString * const kLocalSSLHost = NSStringize(HOST_PORT_LOCALSSL); // The Protocol Buffers encoding overhead of local interop server. Acquired // by experiment. Adjust this when server's proto file changes. diff --git a/src/objective-c/tests/InteropTestsRemote.m b/src/objective-c/tests/InteropTestsRemote.m index ff1193302b..5260fe4570 100644 --- a/src/objective-c/tests/InteropTestsRemote.m +++ b/src/objective-c/tests/InteropTestsRemote.m @@ -36,7 +36,11 @@ #import "InteropTests.h" -static NSString * const kRemoteSSLHost = @"grpc-test.sandbox.googleapis.com"; +// The server address is derived from preprocessor macro, which is +// in turn derived from environment variable of the same name. +#define NSStringize_helper(x) #x +#define NSStringize(x) @NSStringize_helper(x) +static NSString * const kRemoteSSLHost = NSStringize(HOST_PORT_REMOTE); // The Protocol Buffers encoding overhead of remote interop server. Acquired // by experiment. Adjust this when server's proto file changes. diff --git a/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m b/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m index 9edfbc2639..a7f190d2b4 100644 --- a/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m +++ b/src/objective-c/tests/InteropTestsRemoteWithCronet/InteropTestsRemoteWithCronet.m @@ -39,7 +39,12 @@ #import "InteropTests.h" -static NSString * const kRemoteSSLHost = @"grpc-test.sandbox.googleapis.com"; +// The server address is derived from preprocessor macro, which is +// in turn derived from environment variable of the same name. +#define NSStringize_helper(x) #x +#define NSStringize(x) @NSStringize_helper(x) +static NSString * const kRemoteSSLHost = NSStringize(HOST_PORT_REMOTE); + // The Protocol Buffers encoding overhead of remote interop server. Acquired // by experiment. Adjust this when server's proto file changes. diff --git a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj b/src/objective-c/tests/Tests.xcodeproj/project.pbxproj index 97de723a22..b01d5ffcea 100644 --- a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj +++ b/src/objective-c/tests/Tests.xcodeproj/project.pbxproj @@ -1282,6 +1282,9 @@ GCC_PREPROCESSOR_DEFINITIONS = ( "DEBUG=1", "$(inherited)", + "HOST_PORT_LOCALSSL=$(HOST_PORT_LOCALSSL)", + "HOST_PORT_LOCAL=$(HOST_PORT_LOCAL)", + "HOST_PORT_REMOTE=$(HOST_PORT_REMOTE)", ); GCC_SYMBOLS_PRIVATE_EXTERN = NO; GCC_TREAT_WARNINGS_AS_ERRORS = YES; @@ -1565,6 +1568,7 @@ GCC_PREPROCESSOR_DEFINITIONS = ( "DEBUG=1", "$(inherited)", + "HOST_PORT_REMOTE=$(HOST_PORT_REMOTE)", ); GCC_SYMBOLS_PRIVATE_EXTERN = NO; GCC_TREAT_WARNINGS_AS_ERRORS = YES; diff --git a/src/objective-c/tests/run_tests.sh b/src/objective-c/tests/run_tests.sh index 0e82bcaa44..2432209f4f 100755 --- a/src/objective-c/tests/run_tests.sh +++ b/src/objective-c/tests/run_tests.sh @@ -59,6 +59,9 @@ xcodebuild \ -workspace Tests.xcworkspace \ -scheme AllTests \ -destination name="iPhone 6" \ + HOST_PORT_LOCALSSL=localhost:5051 \ + HOST_PORT_LOCAL=localhost:5050 \ + HOST_PORT_REMOTE=grpc-test.sandbox.googleapis.com \ test | xcpretty echo "TIME: $(date)" @@ -84,4 +87,5 @@ xcodebuild \ -workspace Tests.xcworkspace \ -scheme InteropTestsRemoteWithCronet \ -destination name="iPhone 6" \ + HOST_PORT_REMOTE=grpc-test.sandbox.googleapis.com \ test | xcpretty diff --git a/src/proto/grpc/lb/v1/load_balancer.proto b/src/proto/grpc/lb/v1/load_balancer.proto index 44a5150a7e..a2502fb284 100644 --- a/src/proto/grpc/lb/v1/load_balancer.proto +++ b/src/proto/grpc/lb/v1/load_balancer.proto @@ -45,6 +45,20 @@ message Duration { int32 nanos = 2; } +message Timestamp { + + // Represents seconds of UTC time since Unix epoch + // 1970-01-01T00:00:00Z. Must be from 0001-01-01T00:00:00Z to + // 9999-12-31T23:59:59Z inclusive. + int64 seconds = 1; + + // Non-negative fractions of a second at nanosecond resolution. Negative + // second values with fractions must still have non-negative nanos values + // that count forward in time. Must be from 0 to 999,999,999 + // inclusive. + int32 nanos = 2; +} + service LoadBalancer { // Bidirectional rpc to get a list of servers. rpc BalanceLoad(stream LoadBalanceRequest) @@ -63,22 +77,37 @@ message LoadBalanceRequest { } message InitialLoadBalanceRequest { - // Name of load balanced service (IE, service.grpc.gslb.google.com). Its + // Name of load balanced service (IE, balancer.service.com) // length should be less than 256 bytes. string name = 1; } // Contains client level statistics that are useful to load balancing. Each -// count should be reset to zero after reporting the stats. +// count except the timestamp should be reset to zero after reporting the stats. message ClientStats { - // The total number of requests sent by the client since the last report. - int64 total_requests = 1; + // The timestamp of generating the report. + Timestamp timestamp = 1; - // The number of client rpc errors since the last report. - int64 client_rpc_errors = 2; + // The total number of RPCs that started. + int64 num_calls_started = 2; - // The number of dropped requests since the last report. - int64 dropped_requests = 3; + // The total number of RPCs that finished. + int64 num_calls_finished = 3; + + // The total number of RPCs that were dropped by the client because of rate + // limiting. + int64 num_calls_finished_with_drop_for_rate_limiting = 4; + + // The total number of RPCs that were dropped by the client because of load + // balancing. + int64 num_calls_finished_with_drop_for_load_balancing = 5; + + // The total number of RPCs that failed to reach a server except dropped RPCs. + int64 num_calls_finished_with_client_failed_to_send = 6; + + // The total number of RPCs that finished and are known to have been received + // by a server. + int64 num_calls_finished_known_received = 7; } message LoadBalanceResponse { @@ -120,6 +149,10 @@ message ServerList { Duration expiration_interval = 3; } +// Contains server information. When none of the [drop_for_*] fields are true, +// use the other fields. When drop_for_rate_limiting is true, ignore all other +// fields. Use drop_for_load_balancing only when it is true and +// drop_for_rate_limiting is false. message Server { // A resolved address for the server, serialized in network-byte-order. It may // either be an IPv4 or IPv6 address. @@ -137,6 +170,10 @@ message Server { string load_balance_token = 3; // Indicates whether this particular request should be dropped by the client - // when this server is chosen from the list. - bool drop_request = 4; + // for rate limiting. + bool drop_for_rate_limiting = 4; + + // Indicates whether this particular request should be dropped by the client + // for load balancing. + bool drop_for_load_balancing = 5; } diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.c b/src/ruby/ext/grpc/rb_grpc_imports.generated.c index 1d204515cc..1c68b2dc7c 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.c +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.c @@ -111,7 +111,6 @@ grpc_channel_create_registered_call_type grpc_channel_create_registered_call_imp grpc_call_arena_alloc_type grpc_call_arena_alloc_import; grpc_call_start_batch_type grpc_call_start_batch_import; grpc_call_get_peer_type grpc_call_get_peer_import; -grpc_call_set_load_reporting_cost_context_type grpc_call_set_load_reporting_cost_context_import; grpc_census_call_set_context_type grpc_census_call_set_context_import; grpc_census_call_get_context_type grpc_census_call_get_context_import; grpc_channel_get_target_type grpc_channel_get_target_import; @@ -410,7 +409,6 @@ void grpc_rb_load_imports(HMODULE library) { grpc_call_arena_alloc_import = (grpc_call_arena_alloc_type) GetProcAddress(library, "grpc_call_arena_alloc"); grpc_call_start_batch_import = (grpc_call_start_batch_type) GetProcAddress(library, "grpc_call_start_batch"); grpc_call_get_peer_import = (grpc_call_get_peer_type) GetProcAddress(library, "grpc_call_get_peer"); - grpc_call_set_load_reporting_cost_context_import = (grpc_call_set_load_reporting_cost_context_type) GetProcAddress(library, "grpc_call_set_load_reporting_cost_context"); grpc_census_call_set_context_import = (grpc_census_call_set_context_type) GetProcAddress(library, "grpc_census_call_set_context"); grpc_census_call_get_context_import = (grpc_census_call_get_context_type) GetProcAddress(library, "grpc_census_call_get_context"); grpc_channel_get_target_import = (grpc_channel_get_target_type) GetProcAddress(library, "grpc_channel_get_target"); diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h index b691df9ba3..c33fdd210e 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h @@ -284,9 +284,6 @@ extern grpc_call_start_batch_type grpc_call_start_batch_import; typedef char *(*grpc_call_get_peer_type)(grpc_call *call); extern grpc_call_get_peer_type grpc_call_get_peer_import; #define grpc_call_get_peer grpc_call_get_peer_import -typedef void(*grpc_call_set_load_reporting_cost_context_type)(grpc_call *call, struct grpc_load_reporting_cost_context *context); -extern grpc_call_set_load_reporting_cost_context_type grpc_call_set_load_reporting_cost_context_import; -#define grpc_call_set_load_reporting_cost_context grpc_call_set_load_reporting_cost_context_import typedef void(*grpc_census_call_set_context_type)(grpc_call *call, struct census_context *context); extern grpc_census_call_set_context_type grpc_census_call_set_context_import; #define grpc_census_call_set_context grpc_census_call_set_context_import |