diff options
author | 2017-03-28 10:13:07 -0700 | |
---|---|---|
committer | 2017-03-28 10:13:07 -0700 | |
commit | b6090a697b6559155ef8ac80d4343204183571d7 (patch) | |
tree | 1fffa6968d3ee46476a1106e70f3730144412bdb /src/core/lib/channel | |
parent | 306efc787a8a224b4fc5b523fd551f402932d14e (diff) | |
parent | 739cecb0bc1f1ba3b2e0b390795cbaf429ec81c2 (diff) |
Merge github.com:grpc/grpc into new_transport_op
Diffstat (limited to 'src/core/lib/channel')
-rw-r--r-- | src/core/lib/channel/connected_channel.c | 3 | ||||
-rw-r--r-- | src/core/lib/channel/deadline_filter.c | 6 | ||||
-rw-r--r-- | src/core/lib/channel/handshaker.c | 5 | ||||
-rw-r--r-- | src/core/lib/channel/http_client_filter.c | 125 | ||||
-rw-r--r-- | src/core/lib/channel/http_server_filter.c | 111 | ||||
-rw-r--r-- | src/core/lib/channel/message_size_filter.c | 11 |
6 files changed, 171 insertions, 90 deletions
diff --git a/src/core/lib/channel/connected_channel.c b/src/core/lib/channel/connected_channel.c index 42ef7b7806..75c68a5534 100644 --- a/src/core/lib/channel/connected_channel.c +++ b/src/core/lib/channel/connected_channel.c @@ -90,7 +90,8 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, exec_ctx, chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld), &args->call_stack->refcount, args->server_transport_data, args->arena); return r == 0 ? GRPC_ERROR_NONE - : GRPC_ERROR_CREATE("transport stream initialization failed"); + : GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "transport stream initialization failed"); } static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c index 9c484101e8..939ed21677 100644 --- a/src/core/lib/channel/deadline_filter.c +++ b/src/core/lib/channel/deadline_filter.c @@ -55,9 +55,9 @@ static void timer_callback(grpc_exec_ctx* exec_ctx, void* arg, if (error != GRPC_ERROR_CANCELLED) { grpc_call_element_signal_error( exec_ctx, elem, - grpc_error_set_int(GRPC_ERROR_CREATE("Deadline Exceeded"), - GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_DEADLINE_EXCEEDED)); + grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Deadline Exceeded"), + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_DEADLINE_EXCEEDED)); } GRPC_CALL_STACK_UNREF(exec_ctx, deadline_state->call_stack, "deadline_timer"); } diff --git a/src/core/lib/channel/handshaker.c b/src/core/lib/channel/handshaker.c index 1b4240bb10..5861fa6f54 100644 --- a/src/core/lib/channel/handshaker.c +++ b/src/core/lib/channel/handshaker.c @@ -236,8 +236,9 @@ static void call_next_handshaker(grpc_exec_ctx* exec_ctx, void* arg, static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { grpc_handshake_manager* mgr = arg; if (error == GRPC_ERROR_NONE) { // Timer fired, rather than being cancelled. - grpc_handshake_manager_shutdown(exec_ctx, mgr, - GRPC_ERROR_CREATE("Handshake timed out")); + grpc_handshake_manager_shutdown( + exec_ctx, mgr, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Handshake timed out")); } grpc_handshake_manager_unref(exec_ctx, mgr); } diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c index a0cbeaabf3..e43b97335c 100644 --- a/src/core/lib/channel/http_client_filter.c +++ b/src/core/lib/channel/http_client_filter.c @@ -36,6 +36,7 @@ #include <grpc/support/string_util.h> #include <string.h> #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/security/util/b64.h" #include "src/core/lib/slice/percent_encoding.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" @@ -56,7 +57,6 @@ typedef struct call_data { grpc_linked_mdelem te_trailers; grpc_linked_mdelem content_type; grpc_linked_mdelem user_agent; - grpc_linked_mdelem payload_bin; grpc_metadata_batch *recv_initial_metadata; grpc_metadata_batch *recv_trailing_metadata; @@ -108,11 +108,11 @@ static grpc_error *client_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, grpc_error *e = grpc_error_set_str( grpc_error_set_int( grpc_error_set_str( - GRPC_ERROR_CREATE( + GRPC_ERROR_CREATE_FROM_STATIC_STRING( "Received http2 :status header with non-200 OK status"), - GRPC_ERROR_STR_VALUE, val), + GRPC_ERROR_STR_VALUE, grpc_slice_from_copied_string(val)), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_CANCELLED), - GRPC_ERROR_STR_GRPC_MESSAGE, msg); + GRPC_ERROR_STR_GRPC_MESSAGE, grpc_slice_from_copied_string(msg)); gpr_free(val); gpr_free(msg); return e; @@ -252,18 +252,13 @@ static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { } } -typedef struct hc_mutate_op_result { - grpc_error *error; - bool op_stalled; -} hc_mutate_op_result; - -static hc_mutate_op_result hc_mutate_op(grpc_exec_ctx *exec_ctx, - grpc_call_element *elem, - grpc_transport_stream_op *op) { +static grpc_error *hc_mutate_op(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_transport_stream_op *op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; - hc_mutate_op_result result = {.error = GRPC_ERROR_NONE, .op_stalled = false}; + grpc_error *error; if (op->send_initial_metadata) { /* Decide which HTTP VERB to use. We use GET if the request is marked @@ -298,23 +293,63 @@ static hc_mutate_op_result hc_mutate_op(grpc_exec_ctx *exec_ctx, calld->send_length = op->payload->send_message.send_message->length; calld->send_flags = op->payload->send_message.send_message->flags; continue_send_message(exec_ctx, elem); - result.op_stalled = true; if (calld->send_message_blocked == false) { - /* when all the send_message data is available, then create a MDELEM and - append to headers */ - grpc_mdelem payload_bin = grpc_mdelem_from_slices( - exec_ctx, GRPC_MDSTR_GRPC_PAYLOAD_BIN, - grpc_slice_from_copied_buffer( - (const char *)calld->payload_bytes, - op->payload->send_message.send_message->length)); - result.error = grpc_metadata_batch_add_tail( - exec_ctx, op->payload->send_initial_metadata.send_initial_metadata, - &calld->payload_bin, payload_bin); - if (result.error != GRPC_ERROR_NONE) return result; + /* when all the send_message data is available, then modify the path + * MDELEM by appending base64 encoded query to the path */ + const int k_url_safe = 1; + const int k_multi_line = 0; + const unsigned char k_query_separator = '?'; + + grpc_slice path_slice = + GRPC_MDVALUE(op->payload->send_initial_metadata + .send_initial_metadata->idx.named.path->md); + /* sum up individual component's lengths and allocate enough memory to + * hold combined path+query */ + size_t estimated_len = GRPC_SLICE_LENGTH(path_slice); + estimated_len++; /* for the '?' */ + 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 */ + uint8_t *write_ptr = + (uint8_t *)GRPC_SLICE_START_PTR(path_with_query_slice); + uint8_t *original_path = (uint8_t *)GRPC_SLICE_START_PTR(path_slice); + memcpy(write_ptr, original_path, GRPC_SLICE_LENGTH(path_slice)); + write_ptr += GRPC_SLICE_LENGTH(path_slice); + + *write_ptr = k_query_separator; + write_ptr++; /* for the '?' */ + + grpc_base64_encode_core((char *)write_ptr, calld->payload_bytes, + op->payload->send_message.send_message->length, + k_url_safe, k_multi_line); + + /* remove trailing unused memory and add trailing 0 to terminate string + */ + char *t = (char *)GRPC_SLICE_START_PTR(path_with_query_slice); + /* safe to use strlen since base64_encode will always add '\0' */ + size_t path_length = strlen(t) + 1; + *(t + path_length) = '\0'; + path_with_query_slice = + grpc_slice_sub(path_with_query_slice, 0, path_length); + + /* substitute previous path with the new path+query */ + grpc_mdelem mdelem_path_and_query = grpc_mdelem_from_slices( + exec_ctx, GRPC_MDSTR_PATH, path_with_query_slice); + grpc_metadata_batch *b = + op->payload->send_initial_metadata.send_initial_metadata; + error = grpc_metadata_batch_substitute(exec_ctx, b, b->idx.named.path, + mdelem_path_and_query); + if (error != GRPC_ERROR_NONE) return error; + calld->on_complete = op->on_complete; op->on_complete = &calld->hc_on_complete; op->send_message = NULL; + grpc_slice_unref_internal(exec_ctx, path_with_query_slice); } else { /* Not all data is available. Fall back to POST. */ gpr_log(GPR_DEBUG, @@ -342,26 +377,26 @@ static hc_mutate_op_result hc_mutate_op(grpc_exec_ctx *exec_ctx, /* Send : prefixed headers, which have to be before any application layer headers. */ - result.error = grpc_metadata_batch_add_head( + error = grpc_metadata_batch_add_head( exec_ctx, op->payload->send_initial_metadata.send_initial_metadata, &calld->method, method); - if (result.error != GRPC_ERROR_NONE) return result; - result.error = grpc_metadata_batch_add_head( + if (error != GRPC_ERROR_NONE) return error; + error = grpc_metadata_batch_add_head( exec_ctx, op->payload->send_initial_metadata.send_initial_metadata, &calld->scheme, channeld->static_scheme); - if (result.error != GRPC_ERROR_NONE) return result; - result.error = grpc_metadata_batch_add_tail( + if (error != GRPC_ERROR_NONE) return error; + error = grpc_metadata_batch_add_tail( exec_ctx, op->payload->send_initial_metadata.send_initial_metadata, &calld->te_trailers, GRPC_MDELEM_TE_TRAILERS); - if (result.error != GRPC_ERROR_NONE) return result; - result.error = grpc_metadata_batch_add_tail( + if (error != GRPC_ERROR_NONE) return error; + error = grpc_metadata_batch_add_tail( exec_ctx, op->payload->send_initial_metadata.send_initial_metadata, &calld->content_type, GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC); - if (result.error != GRPC_ERROR_NONE) return result; - result.error = grpc_metadata_batch_add_tail( + if (error != GRPC_ERROR_NONE) return error; + error = grpc_metadata_batch_add_tail( exec_ctx, op->payload->send_initial_metadata.send_initial_metadata, &calld->user_agent, GRPC_MDELEM_REF(channeld->user_agent)); - if (result.error != GRPC_ERROR_NONE) return result; + if (error != GRPC_ERROR_NONE) return error; } if (op->recv_initial_metadata) { @@ -382,7 +417,7 @@ static hc_mutate_op_result hc_mutate_op(grpc_exec_ctx *exec_ctx, op->on_complete = &calld->hc_on_recv_trailing_metadata; } - return result; + return GRPC_ERROR_NONE; } static void hc_start_transport_op(grpc_exec_ctx *exec_ctx, @@ -390,14 +425,18 @@ static void hc_start_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) { GPR_TIMER_BEGIN("hc_start_transport_op", 0); GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - hc_mutate_op_result result = hc_mutate_op(exec_ctx, elem, op); - if (result.error != GRPC_ERROR_NONE) { - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, result.error); - } else if (result.op_stalled) { - /* Don't forward the op. send_message contains slices that aren't ready yet. - The call will be forwarded by the op_complete of slice read call. */ + grpc_error *error = hc_mutate_op(exec_ctx, elem, op); + if (error != GRPC_ERROR_NONE) { + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error); } else { - grpc_call_next_op(exec_ctx, elem, op); + call_data *calld = elem->call_data; + if (op->send_message && calld->send_message_blocked) { + /* Don't forward the op. send_message contains slices that aren't ready + yet. The call will be forwarded by the op_complete of slice read call. + */ + } else { + grpc_call_next_op(exec_ctx, elem, op); + } } GPR_TIMER_END("hc_start_transport_op", 0); } diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c index a1ebc8810a..ac4cdcfa97 100644 --- a/src/core/lib/channel/http_server_filter.c +++ b/src/core/lib/channel/http_server_filter.c @@ -37,6 +37,7 @@ #include <grpc/support/log.h> #include <string.h> #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/security/util/b64.h" #include "src/core/lib/slice/percent_encoding.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" @@ -51,8 +52,8 @@ typedef struct call_data { grpc_linked_mdelem status; grpc_linked_mdelem content_type; - /* did this request come with payload-bin */ - bool seen_payload_bin; + /* did this request come with path query containing request payload */ + bool seen_path_with_query; /* flag to ensure payload_bin is delivered only once */ bool payload_bin_delivered; @@ -60,7 +61,7 @@ typedef struct call_data { uint32_t *recv_initial_metadata_flags; /** Closure to call when finished with the hs_on_recv hook */ grpc_closure *on_done_recv; - /** Closure to call when we retrieve read message from the payload-bin header + /** Closure to call when we retrieve read message from the path URI */ grpc_closure *recv_message_ready; grpc_closure *on_complete; @@ -100,7 +101,7 @@ static void add_error(const char *error_name, grpc_error **cumulative, grpc_error *new) { if (new == GRPC_ERROR_NONE) return; if (*cumulative == GRPC_ERROR_NONE) { - *cumulative = GRPC_ERROR_CREATE(error_name); + *cumulative = GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_name); } *cumulative = grpc_error_add_child(*cumulative, new); } @@ -131,27 +132,32 @@ static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, ~GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST; } else { add_error(error_name, &error, - grpc_attach_md_to_error(GRPC_ERROR_CREATE("Bad header"), - b->idx.named.method->md)); + grpc_attach_md_to_error( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Bad header"), + b->idx.named.method->md)); } grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.method); } else { - add_error(error_name, &error, - grpc_error_set_str(GRPC_ERROR_CREATE("Missing header"), - GRPC_ERROR_STR_KEY, ":method")); + add_error( + error_name, &error, + grpc_error_set_str( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Missing header"), + GRPC_ERROR_STR_KEY, grpc_slice_from_static_string(":method"))); } if (b->idx.named.te != NULL) { if (!grpc_mdelem_eq(b->idx.named.te->md, GRPC_MDELEM_TE_TRAILERS)) { add_error(error_name, &error, - grpc_attach_md_to_error(GRPC_ERROR_CREATE("Bad header"), - b->idx.named.te->md)); + grpc_attach_md_to_error( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Bad header"), + b->idx.named.te->md)); } grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.te); } else { add_error(error_name, &error, - grpc_error_set_str(GRPC_ERROR_CREATE("Missing header"), - GRPC_ERROR_STR_KEY, "te")); + grpc_error_set_str( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Missing header"), + GRPC_ERROR_STR_KEY, grpc_slice_from_static_string("te"))); } if (b->idx.named.scheme != NULL) { @@ -159,14 +165,17 @@ static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, !grpc_mdelem_eq(b->idx.named.scheme->md, GRPC_MDELEM_SCHEME_HTTPS) && !grpc_mdelem_eq(b->idx.named.scheme->md, GRPC_MDELEM_SCHEME_GRPC)) { add_error(error_name, &error, - grpc_attach_md_to_error(GRPC_ERROR_CREATE("Bad header"), - b->idx.named.scheme->md)); + grpc_attach_md_to_error( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Bad header"), + b->idx.named.scheme->md)); } grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.scheme); } else { - add_error(error_name, &error, - grpc_error_set_str(GRPC_ERROR_CREATE("Missing header"), - GRPC_ERROR_STR_KEY, ":scheme")); + add_error( + error_name, &error, + grpc_error_set_str( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Missing header"), + GRPC_ERROR_STR_KEY, grpc_slice_from_static_string(":scheme"))); } if (b->idx.named.content_type != NULL) { @@ -200,8 +209,46 @@ static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, if (b->idx.named.path == NULL) { add_error(error_name, &error, - grpc_error_set_str(GRPC_ERROR_CREATE("Missing header"), - GRPC_ERROR_STR_KEY, ":path")); + grpc_error_set_str( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Missing header"), + GRPC_ERROR_STR_KEY, grpc_slice_from_static_string(":path"))); + } else if (*calld->recv_cacheable_request == true) { + /* We have a cacheable request made with GET verb. The path contains the + * query parameter which is base64 encoded request payload. */ + const char k_query_separator = '?'; + grpc_slice path_slice = GRPC_MDVALUE(b->idx.named.path->md); + uint8_t *path_ptr = (uint8_t *)GRPC_SLICE_START_PTR(path_slice); + size_t path_length = GRPC_SLICE_LENGTH(path_slice); + /* offset of the character '?' */ + size_t offset = 0; + for (offset = 0; *path_ptr != k_query_separator && offset < path_length; + path_ptr++, offset++) + ; + if (offset < path_length) { + grpc_slice query_slice = + grpc_slice_sub(path_slice, offset + 1, path_length); + + /* substitute path metadata with just the path (not query) */ + grpc_mdelem mdelem_path_without_query = grpc_mdelem_from_slices( + exec_ctx, GRPC_MDSTR_PATH, grpc_slice_sub(path_slice, 0, offset)); + + grpc_metadata_batch_substitute(exec_ctx, b, b->idx.named.path, + mdelem_path_without_query); + + /* decode payload from query and add to the slice buffer to be returned */ + const int k_url_safe = 1; + grpc_slice_buffer_add( + &calld->read_slice_buffer, + grpc_base64_decode(exec_ctx, + (const char *)GRPC_SLICE_START_PTR(query_slice), + k_url_safe)); + grpc_slice_buffer_stream_init(&calld->read_stream, + &calld->read_slice_buffer, 0); + calld->seen_path_with_query = true; + grpc_slice_unref_internal(exec_ctx, query_slice); + } else { + gpr_log(GPR_ERROR, "GET request without QUERY"); + } } if (b->idx.named.host != NULL && b->idx.named.authority == NULL) { @@ -218,19 +265,11 @@ static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, } if (b->idx.named.authority == NULL) { - add_error(error_name, &error, - grpc_error_set_str(GRPC_ERROR_CREATE("Missing header"), - GRPC_ERROR_STR_KEY, ":authority")); - } - - if (b->idx.named.grpc_payload_bin != NULL) { - calld->seen_payload_bin = true; - grpc_slice_buffer_add(&calld->read_slice_buffer, - grpc_slice_ref_internal( - GRPC_MDVALUE(b->idx.named.grpc_payload_bin->md))); - grpc_slice_buffer_stream_init(&calld->read_stream, - &calld->read_slice_buffer, 0); - grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.grpc_payload_bin); + add_error( + error_name, &error, + grpc_error_set_str( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Missing header"), + GRPC_ERROR_STR_KEY, grpc_slice_from_static_string(":authority"))); } return error; @@ -253,8 +292,8 @@ static void hs_on_complete(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *err) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; - /* Call recv_message_ready if we got the payload via the header field */ - if (calld->seen_payload_bin && calld->recv_message_ready != NULL) { + /* Call recv_message_ready if we got the payload via the path field */ + if (calld->seen_path_with_query && calld->recv_message_ready != NULL) { *calld->pp_recv_message = calld->payload_bin_delivered ? NULL : (grpc_byte_stream *)&calld->read_stream; @@ -269,7 +308,7 @@ static void hs_recv_message_ready(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *err) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; - if (calld->seen_payload_bin) { + if (calld->seen_path_with_query) { /* do nothing. This is probably a GET request, and payload will be returned in hs_on_complete callback. */ } else { diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c index c55b0971e9..0873d9c285 100644 --- a/src/core/lib/channel/message_size_filter.c +++ b/src/core/lib/channel/message_size_filter.c @@ -121,8 +121,8 @@ static void recv_message_ready(grpc_exec_ctx* exec_ctx, void* user_data, "Received message larger than max (%u vs. %d)", (*calld->recv_message)->length, calld->max_recv_size); grpc_error* new_error = grpc_error_set_int( - GRPC_ERROR_CREATE(message_string), GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_INVALID_ARGUMENT); + GRPC_ERROR_CREATE_FROM_COPIED_STRING(message_string), + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_INVALID_ARGUMENT); if (error == GRPC_ERROR_NONE) { error = new_error; } else { @@ -149,9 +149,10 @@ static void start_transport_stream_op(grpc_exec_ctx* exec_ctx, op->payload->send_message.send_message->length, calld->max_send_size); grpc_transport_stream_op_finish_with_failure( - exec_ctx, op, grpc_error_set_int(GRPC_ERROR_CREATE(message_string), - GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_INVALID_ARGUMENT)); + exec_ctx, op, + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_COPIED_STRING(message_string), + GRPC_ERROR_INT_GRPC_STATUS, + GRPC_STATUS_INVALID_ARGUMENT)); gpr_free(message_string); return; } |