diff options
Diffstat (limited to 'src')
101 files changed, 1236 insertions, 703 deletions
diff --git a/src/compiler/ruby_generator.cc b/src/compiler/ruby_generator.cc index 16632325dc..8196589a5c 100644 --- a/src/compiler/ruby_generator.cc +++ b/src/compiler/ruby_generator.cc @@ -67,10 +67,9 @@ void PrintMethod(const MethodDescriptor *method, const std::string &package, if (method->server_streaming()) { output_type = "stream(" + output_type + ")"; } - std::map<std::string, std::string> method_vars = ListToDict({ - "mth.name", method->name(), "input.type", input_type, "output.type", - output_type, - }); + std::map<std::string, std::string> method_vars = + ListToDict({"mth.name", method->name(), "input.type", input_type, + "output.type", output_type, }); out->Print(method_vars, "rpc :$mth.name$, $input.type$, $output.type$\n"); } @@ -82,17 +81,15 @@ void PrintService(const ServiceDescriptor *service, const std::string &package, } // Begin the service module - std::map<std::string, std::string> module_vars = ListToDict({ - "module.name", CapitalizeFirst(service->name()), - }); + std::map<std::string, std::string> module_vars = + ListToDict({"module.name", CapitalizeFirst(service->name()), }); out->Print(module_vars, "module $module.name$\n"); out->Indent(); // TODO(temiola): add documentation std::string doc = "TODO: add proto service documentation here"; - std::map<std::string, std::string> template_vars = ListToDict({ - "Documentation", doc, - }); + std::map<std::string, std::string> template_vars = + ListToDict({"Documentation", doc, }); out->Print("\n"); out->Print(template_vars, "# $Documentation$\n"); out->Print("class Service\n"); @@ -104,9 +101,8 @@ void PrintService(const ServiceDescriptor *service, const std::string &package, out->Print("\n"); out->Print("self.marshal_class_method = :encode\n"); out->Print("self.unmarshal_class_method = :decode\n"); - std::map<std::string, std::string> pkg_vars = ListToDict({ - "service.name", service->name(), "pkg.name", package, - }); + std::map<std::string, std::string> pkg_vars = + ListToDict({"service.name", service->name(), "pkg.name", package, }); out->Print(pkg_vars, "self.service_name = '$pkg.name$.$service.name$'\n"); out->Print("\n"); for (int i = 0; i < service->method_count(); ++i) { @@ -137,9 +133,8 @@ std::string GetServices(const FileDescriptor *file) { } // Write out a file header. - std::map<std::string, std::string> header_comment_vars = ListToDict({ - "file.name", file->name(), "file.package", file->package(), - }); + std::map<std::string, std::string> header_comment_vars = ListToDict( + {"file.name", file->name(), "file.package", file->package(), }); out.Print("# Generated by the protocol buffer compiler. DO NOT EDIT!\n"); out.Print(header_comment_vars, "# Source: $file.name$ for package '$file.package$'\n"); @@ -149,18 +144,16 @@ std::string GetServices(const FileDescriptor *file) { // Write out require statemment to import the separately generated file // that defines the messages used by the service. This is generated by the // main ruby plugin. - std::map<std::string, std::string> dep_vars = ListToDict({ - "dep.name", MessagesRequireName(file), - }); + std::map<std::string, std::string> dep_vars = + ListToDict({"dep.name", MessagesRequireName(file), }); out.Print(dep_vars, "require '$dep.name$'\n"); // Write out services within the modules out.Print("\n"); std::vector<std::string> modules = Split(file->package(), '.'); for (size_t i = 0; i < modules.size(); ++i) { - std::map<std::string, std::string> module_vars = ListToDict({ - "module.name", CapitalizeFirst(modules[i]), - }); + std::map<std::string, std::string> module_vars = + ListToDict({"module.name", CapitalizeFirst(modules[i]), }); out.Print(module_vars, "module $module.name$\n"); out.Indent(); } diff --git a/src/core/channel/call_op_string.c b/src/core/channel/call_op_string.c index d36d51e789..127ea707bf 100644 --- a/src/core/channel/call_op_string.c +++ b/src/core/channel/call_op_string.c @@ -83,6 +83,9 @@ char *grpc_call_op_string(grpc_call_op *op) { case GRPC_SEND_MESSAGE: gpr_strvec_add(&b, gpr_strdup("SEND_MESSAGE")); break; + case GRPC_SEND_PREFORMATTED_MESSAGE: + gpr_strvec_add(&b, gpr_strdup("SEND_PREFORMATTED_MESSAGE")); + break; case GRPC_SEND_FINISH: gpr_strvec_add(&b, gpr_strdup("SEND_FINISH")); break; diff --git a/src/core/channel/census_filter.c b/src/core/channel/census_filter.c index ed60f0a5f6..3447e9dde4 100644 --- a/src/core/channel/census_filter.c +++ b/src/core/channel/census_filter.c @@ -178,19 +178,11 @@ static void destroy_channel_elem(grpc_channel_element* elem) { } const grpc_channel_filter grpc_client_census_filter = { - client_call_op, channel_op, - - sizeof(call_data), client_init_call_elem, client_destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "census-client"}; + client_call_op, channel_op, sizeof(call_data), + client_init_call_elem, client_destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "census-client"}; const grpc_channel_filter grpc_server_census_filter = { - server_call_op, channel_op, - - sizeof(call_data), server_init_call_elem, server_destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "census-server"}; + server_call_op, channel_op, sizeof(call_data), + server_init_call_elem, server_destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "census-server"}; diff --git a/src/core/channel/channel_args.c b/src/core/channel/channel_args.c index 04ce519ff3..5f16c7b7e9 100644 --- a/src/core/channel/channel_args.c +++ b/src/core/channel/channel_args.c @@ -52,7 +52,9 @@ static grpc_arg copy_arg(const grpc_arg *src) { break; case GRPC_ARG_POINTER: dst.value.pointer = src->value.pointer; - dst.value.pointer.p = src->value.pointer.copy(src->value.pointer.p); + dst.value.pointer.p = src->value.pointer.copy + ? src->value.pointer.copy(src->value.pointer.p) + : src->value.pointer.p; break; } return dst; @@ -91,7 +93,9 @@ void grpc_channel_args_destroy(grpc_channel_args *a) { case GRPC_ARG_INTEGER: break; case GRPC_ARG_POINTER: - a->args[i].value.pointer.destroy(a->args[i].value.pointer.p); + if (a->args[i].value.pointer.destroy) { + a->args[i].value.pointer.destroy(a->args[i].value.pointer.p); + } break; } gpr_free(a->args[i].key); diff --git a/src/core/channel/channel_stack.c b/src/core/channel/channel_stack.c index 5ee412bf7d..af47df844a 100644 --- a/src/core/channel/channel_stack.c +++ b/src/core/channel/channel_stack.c @@ -54,7 +54,7 @@ /* Given a size, round up to the next multiple of sizeof(void*) */ #define ROUND_UP_TO_ALIGNMENT_SIZE(x) \ - (((x)+GPR_MAX_ALIGNMENT - 1) & ~(GPR_MAX_ALIGNMENT - 1)) + (((x) + GPR_MAX_ALIGNMENT - 1) & ~(GPR_MAX_ALIGNMENT - 1)) size_t grpc_channel_stack_size(const grpc_channel_filter **filters, size_t filter_count) { @@ -75,9 +75,9 @@ size_t grpc_channel_stack_size(const grpc_channel_filter **filters, return size; } -#define CHANNEL_ELEMS_FROM_STACK(stk) \ - ((grpc_channel_element *)((char *)(stk) + ROUND_UP_TO_ALIGNMENT_SIZE( \ - sizeof(grpc_channel_stack)))) +#define CHANNEL_ELEMS_FROM_STACK(stk) \ + ((grpc_channel_element *)( \ + (char *)(stk) + ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_channel_stack)))) #define CALL_ELEMS_FROM_STACK(stk) \ ((grpc_call_element *)((char *)(stk) + \ @@ -202,6 +202,17 @@ grpc_call_stack *grpc_call_stack_from_top_element(grpc_call_element *elem) { static void do_nothing(void *user_data, grpc_op_error error) {} +void grpc_call_element_recv_metadata(grpc_call_element *cur_elem, + grpc_mdelem *mdelem) { + grpc_call_op metadata_op; + metadata_op.type = GRPC_RECV_METADATA; + metadata_op.dir = GRPC_CALL_UP; + metadata_op.done_cb = do_nothing; + metadata_op.user_data = NULL; + metadata_op.data.metadata = mdelem; + grpc_call_next_op(cur_elem, &metadata_op); +} + void grpc_call_element_send_metadata(grpc_call_element *cur_elem, grpc_mdelem *mdelem) { grpc_call_op metadata_op; @@ -209,7 +220,7 @@ void grpc_call_element_send_metadata(grpc_call_element *cur_elem, metadata_op.dir = GRPC_CALL_DOWN; metadata_op.done_cb = do_nothing; metadata_op.user_data = NULL; - metadata_op.data.metadata = grpc_mdelem_ref(mdelem); + metadata_op.data.metadata = mdelem; grpc_call_next_op(cur_elem, &metadata_op); } @@ -221,3 +232,12 @@ void grpc_call_element_send_cancel(grpc_call_element *cur_elem) { cancel_op.user_data = NULL; grpc_call_next_op(cur_elem, &cancel_op); } + +void grpc_call_element_send_finish(grpc_call_element *cur_elem) { + grpc_call_op cancel_op; + cancel_op.type = GRPC_SEND_FINISH; + cancel_op.dir = GRPC_CALL_DOWN; + cancel_op.done_cb = do_nothing; + cancel_op.user_data = NULL; + grpc_call_next_op(cur_elem, &cancel_op); +} diff --git a/src/core/channel/channel_stack.h b/src/core/channel/channel_stack.h index eb68102b43..ec9ecf3d15 100644 --- a/src/core/channel/channel_stack.h +++ b/src/core/channel/channel_stack.h @@ -69,6 +69,8 @@ typedef enum { GRPC_SEND_START, /* send a message to the channels peer */ GRPC_SEND_MESSAGE, + /* send a pre-formatted message to the channels peer */ + GRPC_SEND_PREFORMATTED_MESSAGE, /* send half-close to the channels peer */ GRPC_SEND_FINISH, /* request that more data be allowed through flow control */ @@ -244,7 +246,9 @@ typedef struct { /* A call stack tracks a set of related filters for one call, and guarantees they live within a single malloc() allocation */ -typedef struct { size_t count; } grpc_call_stack; +typedef struct { + size_t count; +} grpc_call_stack; /* Get a channel element given a channel stack and its index */ grpc_channel_element *grpc_channel_stack_element(grpc_channel_stack *stack, @@ -292,7 +296,10 @@ void grpc_call_log_op(char *file, int line, gpr_log_severity severity, void grpc_call_element_send_metadata(grpc_call_element *cur_elem, grpc_mdelem *elem); +void grpc_call_element_recv_metadata(grpc_call_element *cur_elem, + grpc_mdelem *elem); void grpc_call_element_send_cancel(grpc_call_element *cur_elem); +void grpc_call_element_send_finish(grpc_call_element *cur_elem); #ifdef GRPC_CHANNEL_STACK_TRACE #define GRPC_CALL_LOG_OP(sev, elem, op) grpc_call_log_op(sev, elem, op) diff --git a/src/core/channel/child_channel.c b/src/core/channel/child_channel.c index f400e9b670..a7f06bcdc0 100644 --- a/src/core/channel/child_channel.c +++ b/src/core/channel/child_channel.c @@ -165,14 +165,9 @@ static void lb_destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_child_channel_top_filter = { - lb_call_op, lb_channel_op, - - sizeof(lb_call_data), lb_init_call_elem, lb_destroy_call_elem, - - sizeof(lb_channel_data), lb_init_channel_elem, lb_destroy_channel_elem, - - "child-channel", -}; + lb_call_op, lb_channel_op, sizeof(lb_call_data), + lb_init_call_elem, lb_destroy_call_elem, sizeof(lb_channel_data), + lb_init_channel_elem, lb_destroy_channel_elem, "child-channel", }; /* grpc_child_channel proper */ diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index f9b42db419..bcb024f2ac 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -450,14 +450,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_client_channel_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "client-channel", -}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "client-channel", }; grpc_transport_setup_result grpc_client_channel_transport_setup_complete( grpc_channel_stack *channel_stack, grpc_transport *transport, diff --git a/src/core/channel/connected_channel.c b/src/core/channel/connected_channel.c index 55486ed5c3..d35cede97b 100644 --- a/src/core/channel/connected_channel.c +++ b/src/core/channel/connected_channel.c @@ -69,7 +69,7 @@ typedef struct { /* We perform a small hack to locate transport data alongside the connected channel data in call allocations, to allow everything to be pulled in minimal cache line requests */ -#define TRANSPORT_STREAM_FROM_CALL_DATA(calld) ((grpc_stream *)((calld)+1)) +#define TRANSPORT_STREAM_FROM_CALL_DATA(calld) ((grpc_stream *)((calld) + 1)) #define CALL_DATA_FROM_TRANSPORT_STREAM(transport_stream) \ (((call_data *)(transport_stream)) - 1) @@ -140,6 +140,8 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, grpc_sopb_add_begin_message(&calld->outgoing_sopb, grpc_byte_buffer_length(op->data.message), op->flags); + /* fall-through */ + case GRPC_SEND_PREFORMATTED_MESSAGE: copy_byte_buffer_to_stream_ops(op->data.message, &calld->outgoing_sopb); calld->outgoing_buffer_length_estimate += (5 + grpc_byte_buffer_length(op->data.message)); @@ -257,14 +259,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_connected_channel_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "connected", -}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "connected", }; static gpr_slice alloc_recv_buffer(void *user_data, grpc_transport *transport, grpc_stream *stream, size_t size_hint) { @@ -508,8 +505,7 @@ static void transport_closed(void *user_data, grpc_transport *transport) { const grpc_transport_callbacks connected_channel_transport_callbacks = { alloc_recv_buffer, accept_stream, recv_batch, - transport_goaway, transport_closed, -}; + transport_goaway, transport_closed, }; grpc_transport_setup_result grpc_connected_channel_bind_transport( grpc_channel_stack *channel_stack, grpc_transport *transport) { diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index b139b72795..4735aa939c 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -35,7 +35,9 @@ #include <string.h> #include <grpc/support/log.h> -typedef struct call_data { int sent_headers; } call_data; +typedef struct call_data { + int sent_headers; +} call_data; typedef struct channel_data { grpc_mdelem *te_trailers; @@ -67,8 +69,8 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, /* Send : prefixed headers, which have to be before any application * layer headers. */ calld->sent_headers = 1; - grpc_call_element_send_metadata(elem, channeld->method); - grpc_call_element_send_metadata(elem, channeld->scheme); + grpc_call_element_send_metadata(elem, grpc_mdelem_ref(channeld->method)); + grpc_call_element_send_metadata(elem, grpc_mdelem_ref(channeld->scheme)); } grpc_call_next_op(elem, op); break; @@ -76,12 +78,12 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, if (!calld->sent_headers) { /* Send : prefixed headers, if we haven't already */ calld->sent_headers = 1; - grpc_call_element_send_metadata(elem, channeld->method); - grpc_call_element_send_metadata(elem, channeld->scheme); + grpc_call_element_send_metadata(elem, grpc_mdelem_ref(channeld->method)); + grpc_call_element_send_metadata(elem, grpc_mdelem_ref(channeld->scheme)); } /* Send non : prefixed headers */ - grpc_call_element_send_metadata(elem, channeld->te_trailers); - grpc_call_element_send_metadata(elem, channeld->content_type); + grpc_call_element_send_metadata(elem, grpc_mdelem_ref(channeld->te_trailers)); + grpc_call_element_send_metadata(elem, grpc_mdelem_ref(channeld->content_type)); grpc_call_next_op(elem, op); break; default: @@ -178,10 +180,6 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_http_client_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "http-client"}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "http-client"}; diff --git a/src/core/channel/http_filter.c b/src/core/channel/http_filter.c index 846f7b9713..eaa746ef20 100644 --- a/src/core/channel/http_filter.c +++ b/src/core/channel/http_filter.c @@ -132,10 +132,6 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_http_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "http"}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "http"}; diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c index 19b9606b43..2658a6d42e 100644 --- a/src/core/channel/http_server_filter.c +++ b/src/core/channel/http_server_filter.c @@ -34,29 +34,80 @@ #include "src/core/channel/http_server_filter.h" #include <string.h> +#include <grpc/grpc_http.h> +#include <grpc/support/alloc.h> #include <grpc/support/log.h> +typedef enum { NOT_RECEIVED, POST, GET } known_method_type; + +typedef struct { + grpc_mdelem *path; + grpc_mdelem *content_type; + grpc_byte_buffer *content; +} gettable; + typedef struct call_data { - int sent_status; - int seen_scheme; - int seen_method; - int seen_te_trailers; + known_method_type seen_method; + gpr_uint8 sent_status; + gpr_uint8 seen_scheme; + gpr_uint8 seen_te_trailers; + grpc_mdelem *path; } call_data; typedef struct channel_data { grpc_mdelem *te_trailers; - grpc_mdelem *method; + grpc_mdelem *method_get; + grpc_mdelem *method_post; grpc_mdelem *http_scheme; grpc_mdelem *https_scheme; /* TODO(klempner): Remove this once we stop using it */ grpc_mdelem *grpc_scheme; grpc_mdelem *content_type; - grpc_mdelem *status; + grpc_mdelem *status_ok; + grpc_mdelem *status_not_found; + grpc_mdstr *path_key; + + size_t gettable_count; + gettable *gettables; } channel_data; /* used to silence 'variable not used' warnings */ static void ignore_unused(void *ignored) {} +/* Handle 'GET': not technically grpc, so probably a web browser hitting + us */ +static void payload_done(void *elem, grpc_op_error error) { + if (error == GRPC_OP_OK) { + grpc_call_element_send_finish(elem); + } +} + +static void handle_get(grpc_call_element *elem) { + channel_data *channeld = elem->channel_data; + call_data *calld = elem->call_data; + grpc_call_op op; + size_t i; + + for (i = 0; i < channeld->gettable_count; i++) { + if (channeld->gettables[i].path == calld->path) { + grpc_call_element_send_metadata(elem, + grpc_mdelem_ref(channeld->status_ok)); + grpc_call_element_send_metadata( + elem, grpc_mdelem_ref(channeld->gettables[i].content_type)); + op.type = GRPC_SEND_PREFORMATTED_MESSAGE; + op.dir = GRPC_CALL_DOWN; + op.flags = 0; + op.data.message = channeld->gettables[i].content; + op.done_cb = payload_done; + op.user_data = elem; + grpc_call_next_op(elem, &op); + } + } + grpc_call_element_send_metadata(elem, + grpc_mdelem_ref(channeld->status_not_found)); + grpc_call_element_send_finish(elem); +} + /* Called either: - in response to an API call (or similar) from above, to send something - a network event (or similar) from below, to receive something @@ -73,14 +124,17 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, case GRPC_RECV_METADATA: /* Check if it is one of the headers we care about. */ if (op->data.metadata == channeld->te_trailers || - op->data.metadata == channeld->method || + op->data.metadata == channeld->method_get || + op->data.metadata == channeld->method_post || op->data.metadata == channeld->http_scheme || op->data.metadata == channeld->https_scheme || op->data.metadata == channeld->grpc_scheme || op->data.metadata == channeld->content_type) { /* swallow it */ - if (op->data.metadata == channeld->method) { - calld->seen_method = 1; + if (op->data.metadata == channeld->method_get) { + calld->seen_method = GET; + } else if (op->data.metadata == channeld->method_post) { + calld->seen_method = POST; } else if (op->data.metadata->key == channeld->http_scheme->key) { calld->seen_scheme = 1; } else if (op->data.metadata == channeld->te_trailers) { @@ -108,7 +162,7 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, grpc_mdelem_unref(op->data.metadata); op->done_cb(op->user_data, GRPC_OP_OK); } else if (op->data.metadata->key == channeld->te_trailers->key || - op->data.metadata->key == channeld->method->key || + op->data.metadata->key == channeld->method_post->key || op->data.metadata->key == channeld->http_scheme->key || op->data.metadata->key == channeld->content_type->key) { gpr_log(GPR_ERROR, "Invalid %s: header: '%s'", @@ -120,6 +174,13 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, grpc_mdelem_unref(op->data.metadata); op->done_cb(op->user_data, GRPC_OP_OK); grpc_call_element_send_cancel(elem); + } else if (op->data.metadata->key == channeld->path_key) { + if (calld->path != NULL) { + gpr_log(GPR_ERROR, "Received :path twice"); + grpc_mdelem_unref(calld->path); + } + calld->path = op->data.metadata; + op->done_cb(op->user_data, GRPC_OP_OK); } else { /* pass the event up */ grpc_call_next_op(elem, op); @@ -129,14 +190,21 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, /* Have we seen the required http2 transport headers? (:method, :scheme, content-type, with :path and :authority covered at the channel level right now) */ - if (calld->seen_method && calld->seen_scheme && calld->seen_te_trailers) { + if (calld->seen_method == POST && calld->seen_scheme && + calld->seen_te_trailers && calld->path) { + grpc_call_element_recv_metadata(elem, calld->path); + calld->path = NULL; grpc_call_next_op(elem, op); + } else if (calld->seen_method == GET) { + handle_get(elem); } else { - if (!calld->seen_method) { + if (calld->seen_method == NOT_RECEIVED) { gpr_log(GPR_ERROR, "Missing :method header"); - } else if (!calld->seen_scheme) { + } + if (!calld->seen_scheme) { gpr_log(GPR_ERROR, "Missing :scheme header"); - } else if (!calld->seen_te_trailers) { + } + if (!calld->seen_te_trailers) { gpr_log(GPR_ERROR, "Missing te trailers header"); } /* Error this call out */ @@ -151,7 +219,8 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, if (!calld->sent_status) { calld->sent_status = 1; /* status is reffed by grpc_call_element_send_metadata */ - grpc_call_element_send_metadata(elem, channeld->status); + grpc_call_element_send_metadata(elem, + grpc_mdelem_ref(channeld->status_ok)); } grpc_call_next_op(elem, op); break; @@ -189,9 +258,10 @@ static void init_call_elem(grpc_call_element *elem, ignore_unused(channeld); /* initialize members */ + calld->path = NULL; calld->sent_status = 0; calld->seen_scheme = 0; - calld->seen_method = 0; + calld->seen_method = NOT_RECEIVED; calld->seen_te_trailers = 0; } @@ -201,14 +271,20 @@ static void destroy_call_elem(grpc_call_element *elem) { call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; - ignore_unused(calld); ignore_unused(channeld); + + if (calld->path) { + grpc_mdelem_unref(calld->path); + } } /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, const grpc_channel_args *args, grpc_mdctx *mdctx, int is_first, int is_last) { + size_t i; + size_t gettable_capacity = 0; + /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; @@ -220,13 +296,40 @@ static void init_channel_elem(grpc_channel_element *elem, /* initialize members */ channeld->te_trailers = grpc_mdelem_from_strings(mdctx, "te", "trailers"); - channeld->status = grpc_mdelem_from_strings(mdctx, ":status", "200"); - channeld->method = grpc_mdelem_from_strings(mdctx, ":method", "POST"); + channeld->status_ok = grpc_mdelem_from_strings(mdctx, ":status", "200"); + channeld->status_not_found = + grpc_mdelem_from_strings(mdctx, ":status", "404"); + channeld->method_post = grpc_mdelem_from_strings(mdctx, ":method", "POST"); + channeld->method_get = grpc_mdelem_from_strings(mdctx, ":method", "GET"); channeld->http_scheme = grpc_mdelem_from_strings(mdctx, ":scheme", "http"); channeld->https_scheme = grpc_mdelem_from_strings(mdctx, ":scheme", "https"); channeld->grpc_scheme = grpc_mdelem_from_strings(mdctx, ":scheme", "grpc"); + channeld->path_key = grpc_mdstr_from_string(mdctx, ":path"); channeld->content_type = grpc_mdelem_from_strings(mdctx, "content-type", "application/grpc"); + + /* initialize http download support */ + channeld->gettable_count = 0; + channeld->gettables = NULL; + for (i = 0; i < args->num_args; i++) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_SERVE_OVER_HTTP)) { + gettable *g; + gpr_slice slice; + grpc_http_server_page *p = args->args[i].value.pointer.p; + if (channeld->gettable_count == gettable_capacity) { + gettable_capacity = + GPR_MAX(gettable_capacity * 3 / 2, gettable_capacity + 1); + channeld->gettables = + gpr_realloc(channeld->gettables, gettable_capacity * sizeof(gettable)); + } + g = &channeld->gettables[channeld->gettable_count++]; + g->path = grpc_mdelem_from_strings(mdctx, ":path", p->path); + g->content_type = + grpc_mdelem_from_strings(mdctx, "content-type", p->content_type); + slice = gpr_slice_from_copied_string(p->content); + g->content = grpc_byte_buffer_create(&slice, 1); + } + } } /* Destructor for channel data */ @@ -235,19 +338,18 @@ static void destroy_channel_elem(grpc_channel_element *elem) { channel_data *channeld = elem->channel_data; grpc_mdelem_unref(channeld->te_trailers); - grpc_mdelem_unref(channeld->status); - grpc_mdelem_unref(channeld->method); + grpc_mdelem_unref(channeld->status_ok); + grpc_mdelem_unref(channeld->status_not_found); + grpc_mdelem_unref(channeld->method_post); + grpc_mdelem_unref(channeld->method_get); grpc_mdelem_unref(channeld->http_scheme); grpc_mdelem_unref(channeld->https_scheme); grpc_mdelem_unref(channeld->grpc_scheme); grpc_mdelem_unref(channeld->content_type); + grpc_mdstr_unref(channeld->path_key); } const grpc_channel_filter grpc_http_server_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "http-server"}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "http-server"}; diff --git a/src/core/channel/metadata_buffer.c b/src/core/channel/metadata_buffer.c index 75fd90b707..a21a37ea7d 100644 --- a/src/core/channel/metadata_buffer.c +++ b/src/core/channel/metadata_buffer.c @@ -61,7 +61,7 @@ struct grpc_metadata_buffer_impl { size_t elem_cap; }; -#define ELEMS(buffer) ((qelem *)((buffer)+1)) +#define ELEMS(buffer) ((qelem *)((buffer) + 1)) void grpc_metadata_buffer_init(grpc_metadata_buffer *buffer) { /* start buffer as NULL, indicating no elements */ @@ -152,7 +152,9 @@ size_t grpc_metadata_buffer_count(const grpc_metadata_buffer *buffer) { return *buffer ? (*buffer)->elems : 0; } -typedef struct { grpc_metadata_buffer_impl *impl; } elems_hdr; +typedef struct { + grpc_metadata_buffer_impl *impl; +} elems_hdr; grpc_metadata *grpc_metadata_buffer_extract_elements( grpc_metadata_buffer *buffer) { diff --git a/src/core/channel/noop_filter.c b/src/core/channel/noop_filter.c index b6b3f661f7..d5615f7ae6 100644 --- a/src/core/channel/noop_filter.c +++ b/src/core/channel/noop_filter.c @@ -131,10 +131,6 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_no_op_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "no-op"}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "no-op"}; diff --git a/src/core/iomgr/alarm_heap.c b/src/core/iomgr/alarm_heap.c index 693d26d03f..2b6198336f 100644 --- a/src/core/iomgr/alarm_heap.c +++ b/src/core/iomgr/alarm_heap.c @@ -66,11 +66,11 @@ static void adjust_downwards(grpc_alarm **first, int i, int length, int next_i; if (left_child >= length) break; right_child = left_child + 1; - next_i = right_child < length && - gpr_time_cmp(first[left_child]->deadline, - first[right_child]->deadline) < 0 - ? right_child - : left_child; + next_i = + right_child < length && gpr_time_cmp(first[left_child]->deadline, + first[right_child]->deadline) < 0 + ? right_child + : left_child; if (gpr_time_cmp(t->deadline, first[next_i]->deadline) >= 0) break; first[i] = first[next_i]; first[i]->heap_index = i; diff --git a/src/core/iomgr/pollset_kick_posix.c b/src/core/iomgr/pollset_kick.c index 4386cf5a46..5ee1cef233 100644 --- a/src/core/iomgr/pollset_kick_posix.c +++ b/src/core/iomgr/pollset_kick.c @@ -34,98 +34,74 @@ #include <grpc/support/port_platform.h> #ifdef GPR_POSIX_SOCKET - -#include "src/core/iomgr/pollset_kick_posix.h" +#include "src/core/iomgr/pollset_kick.h" #include <errno.h> #include <string.h> #include <unistd.h> #include "src/core/iomgr/socket_utils_posix.h" +#include "src/core/iomgr/wakeup_fd_posix.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> -/* This implementation is based on a freelist of pipes. */ - -#define GRPC_MAX_CACHED_PIPES 50 -#define GRPC_PIPE_LOW_WATERMARK 25 +/* This implementation is based on a freelist of wakeup fds, with extra logic to + * handle kicks while there is no attached fd. */ -typedef struct grpc_kick_pipe_info { - int pipe_read_fd; - int pipe_write_fd; - struct grpc_kick_pipe_info *next; -} grpc_kick_pipe_info; +#define GRPC_MAX_CACHED_WFDS 50 +#define GRPC_WFD_LOW_WATERMARK 25 -static grpc_kick_pipe_info *pipe_freelist = NULL; -static int pipe_freelist_count = 0; -static gpr_mu pipe_freelist_mu; +static grpc_kick_fd_info *fd_freelist = NULL; +static int fd_freelist_count = 0; +static gpr_mu fd_freelist_mu; -static grpc_kick_pipe_info *allocate_pipe(void) { - grpc_kick_pipe_info *info; - gpr_mu_lock(&pipe_freelist_mu); - if (pipe_freelist != NULL) { - info = pipe_freelist; - pipe_freelist = pipe_freelist->next; - --pipe_freelist_count; +static grpc_kick_fd_info *allocate_wfd(void) { + grpc_kick_fd_info *info; + gpr_mu_lock(&fd_freelist_mu); + if (fd_freelist != NULL) { + info = fd_freelist; + fd_freelist = fd_freelist->next; + --fd_freelist_count; } else { - int pipefd[2]; - /* TODO(klempner): Make this nonfatal */ - GPR_ASSERT(0 == pipe(pipefd)); - GPR_ASSERT(grpc_set_socket_nonblocking(pipefd[0], 1)); - GPR_ASSERT(grpc_set_socket_nonblocking(pipefd[1], 1)); info = gpr_malloc(sizeof(*info)); - info->pipe_read_fd = pipefd[0]; - info->pipe_write_fd = pipefd[1]; + grpc_wakeup_fd_create(&info->wakeup_fd); info->next = NULL; } - gpr_mu_unlock(&pipe_freelist_mu); + gpr_mu_unlock(&fd_freelist_mu); return info; } -static void destroy_pipe(void) { - /* assumes pipe_freelist_mu is held */ - grpc_kick_pipe_info *current = pipe_freelist; - pipe_freelist = pipe_freelist->next; - pipe_freelist_count--; - close(current->pipe_read_fd); - close(current->pipe_write_fd); +static void destroy_wfd(void) { + /* assumes fd_freelist_mu is held */ + grpc_kick_fd_info *current = fd_freelist; + fd_freelist = fd_freelist->next; + fd_freelist_count--; + grpc_wakeup_fd_destroy(¤t->wakeup_fd); gpr_free(current); } -static void free_pipe(grpc_kick_pipe_info *pipe_info) { - gpr_mu_lock(&pipe_freelist_mu); - pipe_info->next = pipe_freelist; - pipe_freelist = pipe_info; - pipe_freelist_count++; - if (pipe_freelist_count > GRPC_MAX_CACHED_PIPES) { - while (pipe_freelist_count > GRPC_PIPE_LOW_WATERMARK) { - destroy_pipe(); +static void free_wfd(grpc_kick_fd_info *fd_info) { + gpr_mu_lock(&fd_freelist_mu); + fd_info->next = fd_freelist; + fd_freelist = fd_info; + fd_freelist_count++; + if (fd_freelist_count > GRPC_MAX_CACHED_WFDS) { + while (fd_freelist_count > GRPC_WFD_LOW_WATERMARK) { + destroy_wfd(); } } - gpr_mu_unlock(&pipe_freelist_mu); -} - -void grpc_pollset_kick_global_init() { - pipe_freelist = NULL; - gpr_mu_init(&pipe_freelist_mu); -} - -void grpc_pollset_kick_global_destroy() { - while (pipe_freelist != NULL) { - destroy_pipe(); - } - gpr_mu_destroy(&pipe_freelist_mu); + gpr_mu_unlock(&fd_freelist_mu); } void grpc_pollset_kick_init(grpc_pollset_kick_state *kick_state) { gpr_mu_init(&kick_state->mu); kick_state->kicked = 0; - kick_state->pipe_info = NULL; + kick_state->fd_info = NULL; } void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state) { gpr_mu_destroy(&kick_state->mu); - GPR_ASSERT(kick_state->pipe_info == NULL); + GPR_ASSERT(kick_state->fd_info == NULL); } int grpc_pollset_kick_pre_poll(grpc_pollset_kick_state *kick_state) { @@ -135,49 +111,43 @@ int grpc_pollset_kick_pre_poll(grpc_pollset_kick_state *kick_state) { gpr_mu_unlock(&kick_state->mu); return -1; } - kick_state->pipe_info = allocate_pipe(); + kick_state->fd_info = allocate_wfd(); gpr_mu_unlock(&kick_state->mu); - return kick_state->pipe_info->pipe_read_fd; + return GRPC_WAKEUP_FD_GET_READ_FD(&kick_state->fd_info->wakeup_fd); } void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state) { - char buf[128]; - int r; - - for (;;) { - r = read(kick_state->pipe_info->pipe_read_fd, buf, sizeof(buf)); - if (r > 0) continue; - if (r == 0) return; - switch (errno) { - case EAGAIN: - return; - case EINTR: - continue; - default: - gpr_log(GPR_ERROR, "error reading pipe: %s", strerror(errno)); - return; - } - } + grpc_wakeup_fd_consume_wakeup(&kick_state->fd_info->wakeup_fd); } void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state) { gpr_mu_lock(&kick_state->mu); - free_pipe(kick_state->pipe_info); - kick_state->pipe_info = NULL; + free_wfd(kick_state->fd_info); + kick_state->fd_info = NULL; gpr_mu_unlock(&kick_state->mu); } void grpc_pollset_kick_kick(grpc_pollset_kick_state *kick_state) { gpr_mu_lock(&kick_state->mu); - if (kick_state->pipe_info != NULL) { - char c = 0; - while (write(kick_state->pipe_info->pipe_write_fd, &c, 1) != 1 && - errno == EINTR) - ; + if (kick_state->fd_info != NULL) { + grpc_wakeup_fd_wakeup(&kick_state->fd_info->wakeup_fd); } else { kick_state->kicked = 1; } gpr_mu_unlock(&kick_state->mu); } -#endif +void grpc_pollset_kick_global_init_fallback_fd(void) { + grpc_wakeup_fd_global_init_force_fallback(); +} + +void grpc_pollset_kick_global_init(void) { + grpc_wakeup_fd_global_init(); +} + +void grpc_pollset_kick_global_destroy(void) { + grpc_wakeup_fd_global_destroy(); +} + + +#endif /* GPR_POSIX_SOCKET */ diff --git a/src/core/iomgr/pollset_kick.h b/src/core/iomgr/pollset_kick.h index 02f3e41433..b224177d31 100644 --- a/src/core/iomgr/pollset_kick.h +++ b/src/core/iomgr/pollset_kick.h @@ -36,9 +36,6 @@ #include <grpc/support/port_platform.h> -/* This is an abstraction around the typical pipe mechanism for waking up a - thread sitting in a poll() style call. */ - #ifdef GPR_POSIX_SOCKET #include "src/core/iomgr/pollset_kick_posix.h" #endif @@ -47,12 +44,19 @@ #include "src/core/iomgr/pollset_kick_windows.h" #endif +/* This is an abstraction around the typical pipe mechanism for waking up a + thread sitting in a poll() style call. */ + void grpc_pollset_kick_global_init(void); void grpc_pollset_kick_global_destroy(void); void grpc_pollset_kick_init(grpc_pollset_kick_state *kick_state); void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state); +/* Guarantees a pure posix implementation rather than a specialized one, if + * applicable. Intended for testing. */ +void grpc_pollset_kick_global_init_fallback_fd(void); + /* Must be called before entering poll(). If return value is -1, this consumed an existing kick. Otherwise the return value is an FD to add to the poll set. */ diff --git a/src/core/iomgr/pollset_kick_posix.h b/src/core/iomgr/pollset_kick_posix.h index bae3b5923a..162ae5bd8f 100644 --- a/src/core/iomgr/pollset_kick_posix.h +++ b/src/core/iomgr/pollset_kick_posix.h @@ -34,14 +34,18 @@ #ifndef __GRPC_INTERNAL_IOMGR_POLLSET_KICK_POSIX_H_ #define __GRPC_INTERNAL_IOMGR_POLLSET_KICK_POSIX_H_ +#include "src/core/iomgr/wakeup_fd_posix.h" #include <grpc/support/sync.h> -struct grpc_kick_pipe_info; +typedef struct grpc_kick_fd_info { + grpc_wakeup_fd_info wakeup_fd; + struct grpc_kick_fd_info *next; +} grpc_kick_fd_info; typedef struct grpc_pollset_kick_state { gpr_mu mu; int kicked; - struct grpc_kick_pipe_info *pipe_info; + struct grpc_kick_fd_info *fd_info; } grpc_pollset_kick_state; -#endif /* __GRPC_INTERNAL_IOMGR_POLLSET_KICK_POSIX_H_ */ +#endif /* __GRPC_INTERNALIOMGR_POLLSET_KICK_POSIX_H_ */ diff --git a/src/core/iomgr/pollset_kick_windows.h b/src/core/iomgr/pollset_kick_windows.h index 243e519dad..1053230a98 100644 --- a/src/core/iomgr/pollset_kick_windows.h +++ b/src/core/iomgr/pollset_kick_windows.h @@ -36,10 +36,10 @@ #include <grpc/support/sync.h> -struct grpc_kick_pipe_info; +struct grpc_kick_fd_info; typedef struct grpc_pollset_kick_state { int unused; } grpc_pollset_kick_state; -#endif /* __GRPC_INTERNAL_IOMGR_POLLSET_KICK_WINDOWS_H_ */ +#endif /* __GRPC_INTERNALIOMGR_POLLSET_KICK_WINDOWS_H_ */ diff --git a/src/core/iomgr/pollset_windows.h b/src/core/iomgr/pollset_windows.h index 53b9ffa5ab..9214b04b17 100644 --- a/src/core/iomgr/pollset_windows.h +++ b/src/core/iomgr/pollset_windows.h @@ -44,8 +44,8 @@ struct grpc_fd; typedef struct grpc_pollset { - gpr_mu mu; - gpr_cv cv; + gpr_mu mu; + gpr_cv cv; } grpc_pollset; #define GRPC_POLLSET_MU(pollset) (&(pollset)->mu) diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c index 2d6c6a73c2..10daf450b7 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -255,8 +255,7 @@ static int add_socket_to_server(grpc_tcp_server *s, int fd, /* append it to the list under a lock */ if (s->nports == s->port_capacity) { s->port_capacity *= 2; - s->ports = - gpr_realloc(s->ports, sizeof(server_port) * s->port_capacity); + s->ports = gpr_realloc(s->ports, sizeof(server_port) * s->port_capacity); } sp = &s->ports[s->nports++]; sp->server = s; diff --git a/src/core/iomgr/wakeup_fd_eventfd.c b/src/core/iomgr/wakeup_fd_eventfd.c new file mode 100644 index 0000000000..3ee7f94128 --- /dev/null +++ b/src/core/iomgr/wakeup_fd_eventfd.c @@ -0,0 +1,82 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/support/port_platform.h> + +#ifdef GPR_LINUX_EVENTFD + +#include <errno.h> +#include <sys/eventfd.h> +#include <unistd.h> + +#include "src/core/iomgr/wakeup_fd_posix.h" +#include <grpc/support/log.h> + +static void eventfd_create(grpc_wakeup_fd_info *fd_info) { + int efd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + /* TODO(klempner): Handle failure more gracefully */ + GPR_ASSERT(efd >= 0); + fd_info->read_fd = efd; + fd_info->write_fd = -1; +} + +static void eventfd_consume(grpc_wakeup_fd_info *fd_info) { + eventfd_t value; + int err; + do { + err = eventfd_read(fd_info->read_fd, &value); + } while (err < 0 && errno == EINTR); +} + +static void eventfd_wakeup(grpc_wakeup_fd_info *fd_info) { + int err; + do { + err = eventfd_write(fd_info->read_fd, 1); + } while (err < 0 && errno == EINTR); +} + +static void eventfd_destroy(grpc_wakeup_fd_info *fd_info) { + close(fd_info->read_fd); +} + +static int eventfd_check_availability(void) { + /* TODO(klempner): Actually check if eventfd is available */ + return 1; +} + +const grpc_wakeup_fd_vtable specialized_wakeup_fd_vtable = { + eventfd_create, eventfd_consume, eventfd_wakeup, eventfd_destroy, + eventfd_check_availability +}; + +#endif /* GPR_LINUX_EVENTFD */ diff --git a/src/core/iomgr/wakeup_fd_nospecial.c b/src/core/iomgr/wakeup_fd_nospecial.c new file mode 100644 index 0000000000..21e8074d50 --- /dev/null +++ b/src/core/iomgr/wakeup_fd_nospecial.c @@ -0,0 +1,53 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +/* + * This is a dummy file to provide an invalid specialized_wakeup_fd_vtable on + * systems without anything better than pipe. + */ + +#include <grpc/support/port_platform.h> + +#ifndef GPR_POSIX_HAS_SPECIAL_WAKEUP_FD + +#include "src/core/iomgr/wakeup_fd.h" + +static int check_availability_invalid(void) { + return 0; +} + +const grpc_wakeup_fd_vtable specialized_wakeup_fd_vtable = { + NULL, NULL, NULL, NULL, check_availability_invalid +}; + +#endif /* GPR_POSIX_HAS_SPECIAL_WAKEUP */ diff --git a/src/core/iomgr/wakeup_fd_pipe.c b/src/core/iomgr/wakeup_fd_pipe.c new file mode 100644 index 0000000000..f36e6eeb9f --- /dev/null +++ b/src/core/iomgr/wakeup_fd_pipe.c @@ -0,0 +1,93 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +/* TODO(klempner): Allow this code to be disabled. */ +#include "src/core/iomgr/wakeup_fd_posix.h" + +#include <errno.h> +#include <string.h> +#include <unistd.h> + +#include "src/core/iomgr/socket_utils_posix.h" +#include <grpc/support/log.h> + +static void pipe_create(grpc_wakeup_fd_info *fd_info) { + int pipefd[2]; + /* TODO(klempner): Make this nonfatal */ + GPR_ASSERT(0 == pipe(pipefd)); + GPR_ASSERT(grpc_set_socket_nonblocking(pipefd[0], 1)); + GPR_ASSERT(grpc_set_socket_nonblocking(pipefd[1], 1)); + fd_info->read_fd = pipefd[0]; + fd_info->write_fd = pipefd[1]; +} + +static void pipe_consume(grpc_wakeup_fd_info *fd_info) { + char buf[128]; + int r; + + for (;;) { + r = read(fd_info->read_fd, buf, sizeof(buf)); + if (r > 0) continue; + if (r == 0) return; + switch (errno) { + case EAGAIN: + return; + case EINTR: + continue; + default: + gpr_log(GPR_ERROR, "error reading pipe: %s", strerror(errno)); + return; + } + } +} + +static void pipe_wakeup(grpc_wakeup_fd_info *fd_info) { + char c = 0; + while (write(fd_info->write_fd, &c, 1) != 1 && errno == EINTR) + ; +} + +static void pipe_destroy(grpc_wakeup_fd_info *fd_info) { + close(fd_info->read_fd); + close(fd_info->write_fd); +} + +static int pipe_check_availability(void) { + /* Assume that pipes are always available. */ + return 1; +} + +const grpc_wakeup_fd_vtable pipe_wakeup_fd_vtable = { + pipe_create, pipe_consume, pipe_wakeup, pipe_destroy, pipe_check_availability +}; + diff --git a/src/core/iomgr/wakeup_fd_pipe.h b/src/core/iomgr/wakeup_fd_pipe.h new file mode 100644 index 0000000000..fc2898f570 --- /dev/null +++ b/src/core/iomgr/wakeup_fd_pipe.h @@ -0,0 +1,41 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef __GRPC_INTERNAL_IOMGR_WAKEUP_FD_PIPE_H_ +#define __GRPC_INTERNAL_IOMGR_WAKEUP_FD_PIPE_H_ + +#include "src/core/iomgr/wakeup_fd_posix.h" + +extern grpc_wakeup_fd_vtable pipe_wakeup_fd_vtable; + +#endif /* __GRPC_INTERNAL_IOMGR_WAKEUP_FD_PIPE_H_ */ diff --git a/src/core/iomgr/wakeup_fd_posix.c b/src/core/iomgr/wakeup_fd_posix.c new file mode 100644 index 0000000000..9107cf37b1 --- /dev/null +++ b/src/core/iomgr/wakeup_fd_posix.c @@ -0,0 +1,70 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/iomgr/wakeup_fd_posix.h" +#include "src/core/iomgr/wakeup_fd_pipe.h" +#include <stddef.h> + +static const grpc_wakeup_fd_vtable *wakeup_fd_vtable = NULL; + +void grpc_wakeup_fd_global_init(void) { + if (specialized_wakeup_fd_vtable.check_availability()) { + wakeup_fd_vtable = &specialized_wakeup_fd_vtable; + } else { + wakeup_fd_vtable = &pipe_wakeup_fd_vtable; + } +} + +void grpc_wakeup_fd_global_init_force_fallback(void) { + wakeup_fd_vtable = &pipe_wakeup_fd_vtable; +} + +void grpc_wakeup_fd_global_destroy(void) { + wakeup_fd_vtable = NULL; +} + +void grpc_wakeup_fd_create(grpc_wakeup_fd_info *fd_info) { + wakeup_fd_vtable->create(fd_info); +} + +void grpc_wakeup_fd_consume_wakeup(grpc_wakeup_fd_info *fd_info) { + wakeup_fd_vtable->consume(fd_info); +} + +void grpc_wakeup_fd_wakeup(grpc_wakeup_fd_info *fd_info) { + wakeup_fd_vtable->wakeup(fd_info); +} + +void grpc_wakeup_fd_destroy(grpc_wakeup_fd_info *fd_info) { + wakeup_fd_vtable->destroy(fd_info); +} diff --git a/src/core/iomgr/wakeup_fd_posix.h b/src/core/iomgr/wakeup_fd_posix.h new file mode 100644 index 0000000000..c2769afb2a --- /dev/null +++ b/src/core/iomgr/wakeup_fd_posix.h @@ -0,0 +1,102 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +/* + * wakeup_fd abstracts the concept of a file descriptor for the purpose of + * waking up a thread in select()/poll()/epoll_wait()/etc. + + * The poll() family of system calls provide a way for a thread to block until + * there is activity on one (or more) of a set of file descriptors. An + * application may wish to wake up this thread to do non file related work. The + * typical way to do this is to add a pipe to the set of file descriptors, then + * write to the pipe to wake up the thread in poll(). + * + * Linux has a lighter weight eventfd specifically designed for this purpose. + * wakeup_fd abstracts the difference between the two. + * + * Setup: + * 1. Before calling anything, call global_init() at least once. + * 1. Call grpc_wakeup_fd_create() to get a wakeup_fd. + * 2. Add the result of GRPC_WAKEUP_FD_FD to the set of monitored file + * descriptors for the poll() style API you are using. Monitor the file + * descriptor for readability. + * 3. To tear down, call grpc_wakeup_fd_destroy(). This closes the underlying + * file descriptor. + * + * Usage: + * 1. To wake up a polling thread, call grpc_wakeup_fd_wakeup() on a wakeup_fd + * it is monitoring. + * 2. If the polling thread was awakened by a wakeup_fd event, call + * grpc_wakeup_fd_consume_wakeup() on it. + */ +#ifndef __GRPC_INTERNAL_IOMGR_WAKEUP_FD_POSIX_H_ +#define __GRPC_INTERNAL_IOMGR_WAKEUP_FD_POSIX_H_ + +typedef struct grpc_wakeup_fd_info grpc_wakeup_fd_info; + +void grpc_wakeup_fd_global_init(void); +void grpc_wakeup_fd_global_destroy(void); + + +void grpc_wakeup_fd_create(grpc_wakeup_fd_info *fd_info); +void grpc_wakeup_fd_consume_wakeup(grpc_wakeup_fd_info *fd_info); +void grpc_wakeup_fd_wakeup(grpc_wakeup_fd_info *fd_info); +void grpc_wakeup_fd_destroy(grpc_wakeup_fd_info *fd_info); + +#define GRPC_WAKEUP_FD_GET_READ_FD(fd_info) ((fd_info)->read_fd) + +/* Force using the fallback implementation. This is intended for testing + * purposes only.*/ +void grpc_wakeup_fd_global_init_force_fallback(void); + +/* Private structures; don't access their fields directly outside of wakeup fd + * code. */ +struct grpc_wakeup_fd_info { + int read_fd; + int write_fd; +}; + +typedef struct grpc_wakeup_fd_vtable { + void (*create)(grpc_wakeup_fd_info *fd_info); + void (*consume)(grpc_wakeup_fd_info *fd_info); + void (*wakeup)(grpc_wakeup_fd_info *fd_info); + void (*destroy)(grpc_wakeup_fd_info *fd_info); + /* Must be called before calling any other functions */ + int (*check_availability)(void); +} grpc_wakeup_fd_vtable; + +/* Defined in some specialized implementation's .c file, or by + * wakeup_fd_nospecial.c if no such implementation exists. */ +extern const grpc_wakeup_fd_vtable specialized_wakeup_fd_vtable; + +#endif /* __GRPC_INTERNAL_IOMGR_WAKEUP_FD_POSIX_H_ */ diff --git a/src/core/security/auth.c b/src/core/security/auth.c index f743b25838..9d0c075bc3 100644 --- a/src/core/security/auth.c +++ b/src/core/security/auth.c @@ -57,7 +57,7 @@ static void on_credentials_metadata(void *user_data, grpc_mdelem **md_elems, grpc_call_element *elem = (grpc_call_element *)user_data; size_t i; for (i = 0; i < num_md; i++) { - grpc_call_element_send_metadata(elem, md_elems[i]); + grpc_call_element_send_metadata(elem, grpc_mdelem_ref(md_elems[i])); } grpc_call_next_op(elem, &((call_data *)elem->call_data)->op); } diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index 8a9ff41e10..3ec874681a 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -118,9 +118,9 @@ grpc_credentials *grpc_credentials_contains_type( /* Exposed for testing only. */ grpc_credentials_status -grpc_oauth2_token_fetcher_credentials_parse_server_response( - const struct grpc_httpcli_response *response, grpc_mdctx *ctx, - grpc_mdelem **token_elem, gpr_timespec *token_lifetime); + grpc_oauth2_token_fetcher_credentials_parse_server_response( + const struct grpc_httpcli_response *response, grpc_mdctx *ctx, + grpc_mdelem **token_elem, gpr_timespec *token_lifetime); /* Simulates an oauth2 token fetch with the specified value for testing. */ grpc_credentials *grpc_fake_oauth2_credentials_create( diff --git a/src/core/statistics/census_tracing.h b/src/core/statistics/census_tracing.h index 604096ba54..f356c9424d 100644 --- a/src/core/statistics/census_tracing.h +++ b/src/core/statistics/census_tracing.h @@ -34,6 +34,10 @@ #ifndef __GRPC_INTERNAL_STATISTICS_CENSUS_TRACING_H_ #define __GRPC_INTERNAL_STATISTICS_CENSUS_TRACING_H_ +#ifdef __cplusplus +extern "C" { +#endif + /* Opaque structure for trace object */ typedef struct trace_obj trace_obj; @@ -56,4 +60,8 @@ void census_internal_unlock_trace_store(void); /* Gets method tag name associated with the input trace object. */ const char* census_get_trace_method_name(const trace_obj* trace); +#ifdef __cplusplus +} +#endif + #endif /* __GRPC_INTERNAL_STATISTICS_CENSUS_TRACING_H_ */ diff --git a/src/core/support/histogram.c b/src/core/support/histogram.c index c5e5608049..425421c5b9 100644 --- a/src/core/support/histogram.c +++ b/src/core/support/histogram.c @@ -186,10 +186,9 @@ static double threshold_for_count_below(gpr_histogram *h, double count_below) { should lie */ lower_bound = bucket_start(h, lower_idx); upper_bound = bucket_start(h, lower_idx + 1); - return GPR_CLAMP(upper_bound - - (upper_bound - lower_bound) * - (count_so_far - count_below) / - h->buckets[lower_idx], + return GPR_CLAMP(upper_bound - (upper_bound - lower_bound) * + (count_so_far - count_below) / + h->buckets[lower_idx], h->min_seen, h->max_seen); } } diff --git a/src/core/support/log_android.c b/src/core/support/log_android.c index 11129e3e06..53c8153593 100644 --- a/src/core/support/log_android.c +++ b/src/core/support/log_android.c @@ -72,17 +72,15 @@ void gpr_default_log(gpr_log_func_args *args) { final_slash = strrchr(args->file, '/'); if (final_slash == NULL) - display_file = file; + display_file = args->file; else display_file = final_slash + 1; - asprintf(&prefix, "%s:%d] %s", display_file, args->line, args->message); + asprintf(&output, "%s:%d] %s", display_file, args->line, args->message); __android_log_write(severity_to_log_priority(args->severity), "GRPC", output); /* allocated by asprintf => use free, not gpr_free */ - free(prefix); - free(suffix); free(output); } diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 14d990df6a..8bb0779bf7 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -203,7 +203,7 @@ struct grpc_call { gpr_refcount internal_refcount; }; -#define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call)+1)) +#define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call) + 1)) #define CALL_FROM_CALL_STACK(call_stack) (((grpc_call *)(call_stack)) - 1) #define CALL_ELEM_FROM_CALL(call, idx) \ grpc_call_stack_element(CALL_STACK_FROM_CALL(call), idx) @@ -863,7 +863,7 @@ static gpr_uint32 decode_status(grpc_mdelem *md) { gpr_uint32 status; void *user_data = grpc_mdelem_get_user_data(md, destroy_status); if (user_data) { - status = ((gpr_uint32)(gpr_intptr) user_data) - STATUS_OFFSET; + status = ((gpr_uint32)(gpr_intptr)user_data) - STATUS_OFFSET; } else { if (!gpr_parse_bytes_to_uint32(grpc_mdstr_as_c_string(md->value), GPR_SLICE_LENGTH(md->value->slice), @@ -880,8 +880,7 @@ void grpc_call_recv_metadata(grpc_call_element *elem, grpc_call_op *op) { grpc_call *call = CALL_FROM_TOP_ELEM(elem); grpc_mdelem *md = op->data.metadata; grpc_mdstr *key = md->key; - gpr_log(GPR_DEBUG, "call %p got metadata %s %s", call, - grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value)); + if (key == grpc_channel_get_status_string(call->channel)) { maybe_set_status_code(call, decode_status(md)); grpc_mdelem_unref(md); @@ -981,3 +980,8 @@ void grpc_call_set_deadline(grpc_call_element *elem, gpr_timespec deadline) { call->have_alarm = 1; grpc_alarm_init(&call->alarm, deadline, call_alarm, call, gpr_now()); } + +grpc_call_stack *grpc_call_get_call_stack(grpc_call *call) { + return CALL_STACK_FROM_CALL(call); +} + diff --git a/src/core/surface/call.h b/src/core/surface/call.h index 01605bb38a..804b387cb1 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -64,6 +64,8 @@ void grpc_call_client_initial_metadata_complete( void grpc_call_set_deadline(grpc_call_element *surface_element, gpr_timespec deadline); +grpc_call_stack *grpc_call_get_call_stack(grpc_call *call); + /* Given the top call_element, get the call object. */ grpc_call *grpc_call_from_top_element(grpc_call_element *surface_element); diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c index 8ef13675fe..a1bcea58dd 100644 --- a/src/core/surface/channel.c +++ b/src/core/surface/channel.c @@ -51,7 +51,7 @@ struct grpc_channel { grpc_mdstr *authority_string; }; -#define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c)+1)) +#define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c) + 1)) grpc_channel *grpc_channel_create_from_filters( const grpc_channel_filter **filters, size_t num_filters, diff --git a/src/core/surface/client.c b/src/core/surface/client.c index fe3a81f1b9..a7c9b902ed 100644 --- a/src/core/surface/client.c +++ b/src/core/surface/client.c @@ -38,9 +38,13 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> -typedef struct { void *unused; } call_data; +typedef struct { + void *unused; +} call_data; -typedef struct { void *unused; } channel_data; +typedef struct { + void *unused; +} channel_data; static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, grpc_call_op *op) { @@ -109,11 +113,6 @@ static void init_channel_elem(grpc_channel_element *elem, static void destroy_channel_elem(grpc_channel_element *elem) {} const grpc_channel_filter grpc_client_surface_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "client", -}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "client", }; diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c index 056c98646b..6098ac78de 100644 --- a/src/core/surface/lame_client.c +++ b/src/core/surface/lame_client.c @@ -42,9 +42,13 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> -typedef struct { void *unused; } call_data; +typedef struct { + void *unused; +} call_data; -typedef struct { grpc_mdelem *message; } channel_data; +typedef struct { + grpc_mdelem *message; +} channel_data; static void do_nothing(void *data, grpc_op_error error) {} @@ -111,14 +115,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } static const grpc_channel_filter lame_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "lame-client", -}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "lame-client", }; grpc_channel *grpc_lame_client_channel_create(void) { static const grpc_channel_filter *filters[] = {&lame_filter}; diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 9585e4e8ea..9a001f4c33 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -411,14 +411,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } static const grpc_channel_filter server_surface_filter = { - call_op, channel_op, - - sizeof(call_data), init_call_elem, destroy_call_elem, - - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - - "server", -}; + call_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "server", }; static void early_terminate_requested_calls(grpc_completion_queue *cq, void **tags, size_t ntags) { diff --git a/src/core/transport/chttp2/frame_data.c b/src/core/transport/chttp2/frame_data.c index dee61cee50..c4ad8f11d5 100644 --- a/src/core/transport/chttp2/frame_data.c +++ b/src/core/transport/chttp2/frame_data.c @@ -105,28 +105,28 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse( } /* fallthrough */ case GRPC_CHTTP2_DATA_FH_1: - p->frame_size = ((gpr_uint32)*cur) << 24; + p->frame_size = ((gpr_uint32) * cur) << 24; if (++cur == end) { p->state = GRPC_CHTTP2_DATA_FH_2; return GRPC_CHTTP2_PARSE_OK; } /* fallthrough */ case GRPC_CHTTP2_DATA_FH_2: - p->frame_size |= ((gpr_uint32)*cur) << 16; + p->frame_size |= ((gpr_uint32) * cur) << 16; if (++cur == end) { p->state = GRPC_CHTTP2_DATA_FH_3; return GRPC_CHTTP2_PARSE_OK; } /* fallthrough */ case GRPC_CHTTP2_DATA_FH_3: - p->frame_size |= ((gpr_uint32)*cur) << 8; + p->frame_size |= ((gpr_uint32) * cur) << 8; if (++cur == end) { p->state = GRPC_CHTTP2_DATA_FH_4; return GRPC_CHTTP2_PARSE_OK; } /* fallthrough */ case GRPC_CHTTP2_DATA_FH_4: - p->frame_size |= ((gpr_uint32)*cur); + p->frame_size |= ((gpr_uint32) * cur); p->state = GRPC_CHTTP2_DATA_FRAME; ++cur; state->need_flush_reads = 1; diff --git a/src/core/transport/chttp2/frame_goaway.c b/src/core/transport/chttp2/frame_goaway.c index de7c0b010b..3d6e943193 100644 --- a/src/core/transport/chttp2/frame_goaway.c +++ b/src/core/transport/chttp2/frame_goaway.c @@ -75,7 +75,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_LSI0; return GRPC_CHTTP2_PARSE_OK; } - p->last_stream_id = ((gpr_uint32)*cur) << 24; + p->last_stream_id = ((gpr_uint32) * cur) << 24; ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_LSI1: @@ -83,7 +83,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_LSI1; return GRPC_CHTTP2_PARSE_OK; } - p->last_stream_id |= ((gpr_uint32)*cur) << 16; + p->last_stream_id |= ((gpr_uint32) * cur) << 16; ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_LSI2: @@ -91,7 +91,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_LSI2; return GRPC_CHTTP2_PARSE_OK; } - p->last_stream_id |= ((gpr_uint32)*cur) << 8; + p->last_stream_id |= ((gpr_uint32) * cur) << 8; ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_LSI3: @@ -99,7 +99,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_LSI3; return GRPC_CHTTP2_PARSE_OK; } - p->last_stream_id |= ((gpr_uint32)*cur); + p->last_stream_id |= ((gpr_uint32) * cur); ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_ERR0: @@ -107,7 +107,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_ERR0; return GRPC_CHTTP2_PARSE_OK; } - p->error_code = ((gpr_uint32)*cur) << 24; + p->error_code = ((gpr_uint32) * cur) << 24; ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_ERR1: @@ -115,7 +115,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_ERR1; return GRPC_CHTTP2_PARSE_OK; } - p->error_code |= ((gpr_uint32)*cur) << 16; + p->error_code |= ((gpr_uint32) * cur) << 16; ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_ERR2: @@ -123,7 +123,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_ERR2; return GRPC_CHTTP2_PARSE_OK; } - p->error_code |= ((gpr_uint32)*cur) << 8; + p->error_code |= ((gpr_uint32) * cur) << 8; ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_ERR3: @@ -131,7 +131,7 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( p->state = GRPC_CHTTP2_GOAWAY_ERR3; return GRPC_CHTTP2_PARSE_OK; } - p->error_code |= ((gpr_uint32)*cur); + p->error_code |= ((gpr_uint32) * cur); ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_DEBUG: diff --git a/src/core/transport/chttp2/frame_settings.c b/src/core/transport/chttp2/frame_settings.c index 3ca973c07b..63c21a9b92 100644 --- a/src/core/transport/chttp2/frame_settings.c +++ b/src/core/transport/chttp2/frame_settings.c @@ -53,8 +53,7 @@ const grpc_chttp2_setting_parameters {"MAX_FRAME_SIZE", 16384, 16384, 16777215, GRPC_CHTTP2_DISCONNECT_ON_INVALID_VALUE}, {"MAX_HEADER_LIST_SIZE", 0xffffffffu, 0, 0xffffffffu, - GRPC_CHTTP2_CLAMP_INVALID_VALUE}, -}; + GRPC_CHTTP2_CLAMP_INVALID_VALUE}, }; static gpr_uint8 *fill_header(gpr_uint8 *out, gpr_uint32 length, gpr_uint8 flags) { @@ -156,7 +155,7 @@ grpc_chttp2_parse_error grpc_chttp2_settings_parser_parse( } return GRPC_CHTTP2_PARSE_OK; } - parser->id = ((gpr_uint16)*cur) << 8; + parser->id = ((gpr_uint16) * cur) << 8; cur++; /* fallthrough */ case GRPC_CHTTP2_SPS_ID1: @@ -172,7 +171,7 @@ grpc_chttp2_parse_error grpc_chttp2_settings_parser_parse( parser->state = GRPC_CHTTP2_SPS_VAL0; return GRPC_CHTTP2_PARSE_OK; } - parser->value = ((gpr_uint32)*cur) << 24; + parser->value = ((gpr_uint32) * cur) << 24; cur++; /* fallthrough */ case GRPC_CHTTP2_SPS_VAL1: @@ -180,7 +179,7 @@ grpc_chttp2_parse_error grpc_chttp2_settings_parser_parse( parser->state = GRPC_CHTTP2_SPS_VAL1; return GRPC_CHTTP2_PARSE_OK; } - parser->value |= ((gpr_uint32)*cur) << 16; + parser->value |= ((gpr_uint32) * cur) << 16; cur++; /* fallthrough */ case GRPC_CHTTP2_SPS_VAL2: @@ -188,7 +187,7 @@ grpc_chttp2_parse_error grpc_chttp2_settings_parser_parse( parser->state = GRPC_CHTTP2_SPS_VAL2; return GRPC_CHTTP2_PARSE_OK; } - parser->value |= ((gpr_uint32)*cur) << 8; + parser->value |= ((gpr_uint32) * cur) << 8; cur++; /* fallthrough */ case GRPC_CHTTP2_SPS_VAL3: diff --git a/src/core/transport/chttp2/frame_window_update.c b/src/core/transport/chttp2/frame_window_update.c index f61714f52b..04bc690108 100644 --- a/src/core/transport/chttp2/frame_window_update.c +++ b/src/core/transport/chttp2/frame_window_update.c @@ -81,7 +81,7 @@ grpc_chttp2_parse_error grpc_chttp2_window_update_parser_parse( grpc_chttp2_window_update_parser *p = parser; while (p->byte != 4 && cur != end) { - p->amount |= ((gpr_uint32)*cur) << (8 * (3 - p->byte)); + p->amount |= ((gpr_uint32) * cur) << (8 * (3 - p->byte)); cur++; p->byte++; } diff --git a/src/core/transport/chttp2/gen_hpack_tables.c b/src/core/transport/chttp2/gen_hpack_tables.c index cd78fcc39a..fefaf159a5 100644 --- a/src/core/transport/chttp2/gen_hpack_tables.c +++ b/src/core/transport/chttp2/gen_hpack_tables.c @@ -55,21 +55,19 @@ typedef struct { unsigned char index; } spec; -static const spec fields[] = { - {"INDEXED_FIELD", 0X80, 1, 1}, - {"INDEXED_FIELD_X", 0X80, 1, 2}, - {"LITHDR_INCIDX", 0X40, 2, 1}, - {"LITHDR_INCIDX_X", 0X40, 2, 2}, - {"LITHDR_INCIDX_V", 0X40, 2, 0}, - {"LITHDR_NOTIDX", 0X00, 4, 1}, - {"LITHDR_NOTIDX_X", 0X00, 4, 2}, - {"LITHDR_NOTIDX_V", 0X00, 4, 0}, - {"LITHDR_NVRIDX", 0X10, 4, 1}, - {"LITHDR_NVRIDX_X", 0X10, 4, 2}, - {"LITHDR_NVRIDX_V", 0X10, 4, 0}, - {"MAX_TBL_SIZE", 0X20, 3, 1}, - {"MAX_TBL_SIZE_X", 0X20, 3, 2}, -}; +static const spec fields[] = {{"INDEXED_FIELD", 0X80, 1, 1}, + {"INDEXED_FIELD_X", 0X80, 1, 2}, + {"LITHDR_INCIDX", 0X40, 2, 1}, + {"LITHDR_INCIDX_X", 0X40, 2, 2}, + {"LITHDR_INCIDX_V", 0X40, 2, 0}, + {"LITHDR_NOTIDX", 0X00, 4, 1}, + {"LITHDR_NOTIDX_X", 0X00, 4, 2}, + {"LITHDR_NOTIDX_V", 0X00, 4, 0}, + {"LITHDR_NVRIDX", 0X10, 4, 1}, + {"LITHDR_NVRIDX_X", 0X10, 4, 2}, + {"LITHDR_NVRIDX_V", 0X10, 4, 0}, + {"MAX_TBL_SIZE", 0X20, 3, 1}, + {"MAX_TBL_SIZE_X", 0X20, 3, 2}, }; static const int num_fields = sizeof(fields) / sizeof(*fields); @@ -131,9 +129,13 @@ static void generate_first_byte_lut(void) { #define MAXHUFFSTATES 1024 /* represents a set of symbols as an array of booleans indicating inclusion */ -typedef struct { char included[GRPC_CHTTP2_NUM_HUFFSYMS]; } symset; +typedef struct { + char included[GRPC_CHTTP2_NUM_HUFFSYMS]; +} symset; /* represents a lookup table indexed by a nibble */ -typedef struct { int values[16]; } nibblelut; +typedef struct { + int values[16]; +} nibblelut; /* returns a symset that includes all possible symbols */ static symset symset_all(void) { diff --git a/src/core/transport/chttp2/hpack_parser.c b/src/core/transport/chttp2/hpack_parser.c index c98b90e5d1..1eba4a2431 100644 --- a/src/core/transport/chttp2/hpack_parser.c +++ b/src/core/transport/chttp2/hpack_parser.c @@ -221,8 +221,7 @@ static const gpr_uint8 first_byte_lut[256] = { INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, - INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD_X, -}; + INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD, INDEXED_FIELD_X, }; /* state table for huffman decoding: given a state, gives an index/16 into next_sub_tbl. Taking that index and adding the value of the nibble being @@ -242,8 +241,7 @@ static const gpr_uint8 next_tbl[256] = { 38, 1, 1, 1, 1, 1, 1, 1, 15, 2, 2, 2, 2, 26, 3, 3, 39, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 7, 3, 3, 3, 40, 2, 41, 1, 1, 1, 42, 43, 1, 1, 44, 1, 1, 1, 1, 15, 2, 2, 2, 2, 2, 2, - 3, 3, 3, 45, 46, 1, 1, 2, 2, 2, 35, 3, 3, 18, 47, 2, -}; + 3, 3, 3, 45, 46, 1, 1, 2, 2, 2, 35, 3, 3, 18, 47, 2, }; /* next state, based upon current state and the current nibble: see above. generated by gen_hpack_tables.c */ static const gpr_int16 next_sub_tbl[48 * 16] = { @@ -298,8 +296,7 @@ static const gpr_int16 next_sub_tbl[48 * 16] = { 4, 8, 4, 8, 4, 8, 4, 8, 4, 8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 245, 246, 247, 248, 249, 250, 251, 252, 253, 254, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 255, -}; + 0, 0, 255, }; /* emission table: indexed like next_tbl, ultimately gives the byte to be emitted, or -1 for no byte, or 256 for end of stream @@ -322,8 +319,7 @@ static const gpr_uint16 emit_tbl[256] = { 204, 205, 206, 207, 208, 209, 210, 211, 212, 213, 214, 215, 216, 217, 218, 219, 220, 221, 0, 222, 223, 224, 225, 226, 227, 228, 229, 230, 231, 232, 233, 234, 235, 236, 237, 238, 239, 240, 241, 242, 243, 244, 245, 246, 247, - 248, -}; + 248, }; /* generated by gen_hpack_tables.c */ static const gpr_int16 emit_sub_tbl[249 * 16] = { -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, @@ -591,8 +587,7 @@ static const gpr_int16 emit_sub_tbl[249 * 16] = { 251, 251, 252, 252, 253, 253, 254, 254, 2, 3, 4, 5, 6, 7, 8, 11, 12, 14, 15, 16, 17, 18, 19, 20, 21, 23, 24, 25, 26, 27, 28, 29, 30, 31, 127, 220, 249, -1, 10, 10, 10, 10, 13, 13, 13, - 13, 22, 22, 22, 22, 256, 256, 256, 256, -}; + 13, 22, 22, 22, 22, 256, 256, 256, 256, }; static const gpr_uint8 inverse_base64[256] = { 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, @@ -612,8 +607,7 @@ static const gpr_uint8 inverse_base64[256] = { 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, - 255, -}; + 255, }; /* emission helpers */ static void on_hdr(grpc_chttp2_hpack_parser *p, grpc_mdelem *md, @@ -951,7 +945,7 @@ static int parse_value1(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, return 1; } - *p->parsing.value += (((gpr_uint32)*cur) & 0x7f) << 7; + *p->parsing.value += (((gpr_uint32) * cur) & 0x7f) << 7; if ((*cur) & 0x80) { return parse_value2(p, cur + 1, end); @@ -969,7 +963,7 @@ static int parse_value2(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, return 1; } - *p->parsing.value += (((gpr_uint32)*cur) & 0x7f) << 14; + *p->parsing.value += (((gpr_uint32) * cur) & 0x7f) << 14; if ((*cur) & 0x80) { return parse_value3(p, cur + 1, end); @@ -987,7 +981,7 @@ static int parse_value3(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, return 1; } - *p->parsing.value += (((gpr_uint32)*cur) & 0x7f) << 21; + *p->parsing.value += (((gpr_uint32) * cur) & 0x7f) << 21; if ((*cur) & 0x80) { return parse_value4(p, cur + 1, end); @@ -1212,7 +1206,7 @@ static int huff_nibble(grpc_chttp2_hpack_parser *p, gpr_uint8 nibble) { gpr_int16 next = next_sub_tbl[16 * next_tbl[p->huff_state] + nibble]; if (emit != -1) { if (emit >= 0 && emit < 256) { - gpr_uint8 c = (gpr_uint8) emit; + gpr_uint8 c = (gpr_uint8)emit; if (!append_string(p, &c, (&c) + 1)) return 0; } else { assert(emit == 256); diff --git a/src/core/transport/chttp2/hpack_table.c b/src/core/transport/chttp2/hpack_table.c index 8f2ebecfeb..1b944232d8 100644 --- a/src/core/transport/chttp2/hpack_table.c +++ b/src/core/transport/chttp2/hpack_table.c @@ -104,8 +104,7 @@ static struct { /* 58: */ {"user-agent", ""}, /* 59: */ {"vary", ""}, /* 60: */ {"via", ""}, - /* 61: */ {"www-authenticate", ""}, -}; + /* 61: */ {"www-authenticate", ""}, }; void grpc_chttp2_hptbl_init(grpc_chttp2_hptbl *tbl, grpc_mdctx *mdctx) { size_t i; diff --git a/src/core/transport/chttp2/huffsyms.c b/src/core/transport/chttp2/huffsyms.c index 0f86f1bc30..1014a29b9f 100644 --- a/src/core/transport/chttp2/huffsyms.c +++ b/src/core/transport/chttp2/huffsyms.c @@ -293,5 +293,4 @@ const grpc_chttp2_huffsym grpc_chttp2_huffsyms[GRPC_CHTTP2_NUM_HUFFSYMS] = { {0x7ffffef, 27}, {0x7fffff0, 27}, {0x3ffffee, 26}, - {0x3fffffff, 30}, -}; + {0x3fffffff, 30}, }; diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c index 92a36d0c16..c4e3ca516d 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/stream_encoder.c @@ -43,7 +43,7 @@ #include "src/core/transport/chttp2/timeout_encoding.h" #include "src/core/transport/chttp2/varint.h" -#define HASH_FRAGMENT_1(x) ((x)&255) +#define HASH_FRAGMENT_1(x) ((x) & 255) #define HASH_FRAGMENT_2(x) ((x >> 8) & 255) #define HASH_FRAGMENT_3(x) ((x >> 16) & 255) #define HASH_FRAGMENT_4(x) ((x >> 24) & 255) diff --git a/src/core/transport/chttp2/timeout_encoding.c b/src/core/transport/chttp2/timeout_encoding.c index 23c4554cf2..31018c3d27 100644 --- a/src/core/transport/chttp2/timeout_encoding.c +++ b/src/core/transport/chttp2/timeout_encoding.c @@ -60,7 +60,7 @@ static void enc_tiny(char *buffer) { memcpy(buffer, "1n", 3); } static void enc_ext(char *buffer, long value, char ext) { int n = gpr_ltoa(value, buffer); buffer[n] = ext; - buffer[n+1] = 0; + buffer[n + 1] = 0; } static void enc_seconds(char *buffer, long sec) { diff --git a/src/core/transport/chttp2/varint.h b/src/core/transport/chttp2/varint.h index 55f92af3d6..d75869866a 100644 --- a/src/core/transport/chttp2/varint.h +++ b/src/core/transport/chttp2/varint.h @@ -56,7 +56,7 @@ void grpc_chttp2_hpack_write_varint_tail(gpr_uint32 tail_value, ((n) < GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits) \ ? 1 \ : grpc_chttp2_hpack_varint_length( \ - (n)-GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits))) + (n) - GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits))) #define GRPC_CHTTP2_WRITE_VARINT(n, prefix_bits, prefix_or, target, length) \ do { \ @@ -66,7 +66,8 @@ void grpc_chttp2_hpack_write_varint_tail(gpr_uint32 tail_value, } else { \ (tgt)[0] = (prefix_or) | GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits); \ grpc_chttp2_hpack_write_varint_tail( \ - (n)-GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits), (tgt)+1, (length)-1); \ + (n) - GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits), (tgt) + 1, \ + (length) - 1); \ } \ } while (0) diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 531a53b984..5465d3306a 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -525,7 +525,7 @@ static int init_stream(grpc_transport *gt, grpc_stream *gs, lock(t); s->id = 0; } else { - s->id = (gpr_uint32)(gpr_uintptr) server_data; + s->id = (gpr_uint32)(gpr_uintptr)server_data; t->incoming_stream = s; grpc_chttp2_stream_map_add(&t->stream_map, s->id, s); } @@ -1238,7 +1238,7 @@ static int init_header_frame_parser(transport *t, int is_continuation) { t->incoming_stream = NULL; /* if stream is accepted, we set incoming_stream in init_stream */ t->cb->accept_stream(t->cb_user_data, &t->base, - (void *)(gpr_uintptr) t->incoming_stream_id); + (void *)(gpr_uintptr)t->incoming_stream_id); s = t->incoming_stream; if (!s) { gpr_log(GPR_ERROR, "stream not accepted"); @@ -1503,8 +1503,8 @@ static int process_read(transport *t, gpr_slice slice) { "Connect string mismatch: expected '%c' (%d) got '%c' (%d) " "at byte %d", CLIENT_CONNECT_STRING[t->deframe_state], - (int)(gpr_uint8) CLIENT_CONNECT_STRING[t->deframe_state], - *cur, (int)*cur, t->deframe_state); + (int)(gpr_uint8)CLIENT_CONNECT_STRING[t->deframe_state], *cur, + (int)*cur, t->deframe_state); drop_connection(t); return 0; } @@ -1518,7 +1518,7 @@ static int process_read(transport *t, gpr_slice slice) { dts_fh_0: case DTS_FH_0: GPR_ASSERT(cur < end); - t->incoming_frame_size = ((gpr_uint32)*cur) << 16; + t->incoming_frame_size = ((gpr_uint32) * cur) << 16; if (++cur == end) { t->deframe_state = DTS_FH_1; return 1; @@ -1526,7 +1526,7 @@ static int process_read(transport *t, gpr_slice slice) { /* fallthrough */ case DTS_FH_1: GPR_ASSERT(cur < end); - t->incoming_frame_size |= ((gpr_uint32)*cur) << 8; + t->incoming_frame_size |= ((gpr_uint32) * cur) << 8; if (++cur == end) { t->deframe_state = DTS_FH_2; return 1; @@ -1558,7 +1558,7 @@ static int process_read(transport *t, gpr_slice slice) { /* fallthrough */ case DTS_FH_5: GPR_ASSERT(cur < end); - t->incoming_stream_id = (((gpr_uint32)*cur) << 24) & 0x7f; + t->incoming_stream_id = (((gpr_uint32) * cur) << 24) & 0x7f; if (++cur == end) { t->deframe_state = DTS_FH_6; return 1; @@ -1566,7 +1566,7 @@ static int process_read(transport *t, gpr_slice slice) { /* fallthrough */ case DTS_FH_6: GPR_ASSERT(cur < end); - t->incoming_stream_id |= ((gpr_uint32)*cur) << 16; + t->incoming_stream_id |= ((gpr_uint32) * cur) << 16; if (++cur == end) { t->deframe_state = DTS_FH_7; return 1; @@ -1574,7 +1574,7 @@ static int process_read(transport *t, gpr_slice slice) { /* fallthrough */ case DTS_FH_7: GPR_ASSERT(cur < end); - t->incoming_stream_id |= ((gpr_uint32)*cur) << 8; + t->incoming_stream_id |= ((gpr_uint32) * cur) << 8; if (++cur == end) { t->deframe_state = DTS_FH_8; return 1; @@ -1582,7 +1582,7 @@ static int process_read(transport *t, gpr_slice slice) { /* fallthrough */ case DTS_FH_8: GPR_ASSERT(cur < end); - t->incoming_stream_id |= ((gpr_uint32)*cur); + t->incoming_stream_id |= ((gpr_uint32) * cur); t->deframe_state = DTS_FRAME; if (!init_frame_parser(t)) { return 0; diff --git a/src/core/tsi/fake_transport_security.c b/src/core/tsi/fake_transport_security.c index 756b2173ec..0b245e4043 100644 --- a/src/core/tsi/fake_transport_security.c +++ b/src/core/tsi/fake_transport_security.c @@ -369,8 +369,7 @@ static void fake_protector_destroy(tsi_frame_protector* self) { static const tsi_frame_protector_vtable frame_protector_vtable = { fake_protector_protect, fake_protector_protect_flush, - fake_protector_unprotect, fake_protector_destroy, -}; + fake_protector_unprotect, fake_protector_destroy, }; /* --- tsi_handshaker methods implementation. ---*/ @@ -485,8 +484,7 @@ static const tsi_handshaker_vtable handshaker_vtable = { fake_handshaker_get_result, fake_handshaker_extract_peer, fake_handshaker_create_frame_protector, - fake_handshaker_destroy, -}; + fake_handshaker_destroy, }; tsi_handshaker* tsi_create_fake_handshaker(int is_client) { tsi_fake_handshaker* impl = calloc(1, sizeof(tsi_fake_handshaker)); diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c index 1159254a8c..8801cc43df 100644 --- a/src/core/tsi/ssl_transport_security.c +++ b/src/core/tsi/ssl_transport_security.c @@ -703,8 +703,7 @@ static void ssl_protector_destroy(tsi_frame_protector* self) { static const tsi_frame_protector_vtable frame_protector_vtable = { ssl_protector_protect, ssl_protector_protect_flush, ssl_protector_unprotect, - ssl_protector_destroy, -}; + ssl_protector_destroy, }; /* --- tsi_handshaker methods implementation. ---*/ @@ -877,8 +876,7 @@ static const tsi_handshaker_vtable handshaker_vtable = { ssl_handshaker_get_result, ssl_handshaker_extract_peer, ssl_handshaker_create_frame_protector, - ssl_handshaker_destroy, -}; + ssl_handshaker_destroy, }; /* --- tsi_ssl_handshaker_factory common methods. --- */ diff --git a/src/node/binding.gyp b/src/node/binding.gyp index fe4b5da9c8..cf2a6acb04 100644 --- a/src/node/binding.gyp +++ b/src/node/binding.gyp @@ -28,17 +28,17 @@ }, "target_name": "grpc", "sources": [ - "byte_buffer.cc", - "call.cc", - "channel.cc", - "completion_queue_async_worker.cc", - "credentials.cc", - "event.cc", - "node_grpc.cc", - "server.cc", - "server_credentials.cc", - "tag.cc", - "timeval.cc" + "ext/byte_buffer.cc", + "ext/call.cc", + "ext/channel.cc", + "ext/completion_queue_async_worker.cc", + "ext/credentials.cc", + "ext/event.cc", + "ext/node_grpc.cc", + "ext/server.cc", + "ext/server_credentials.cc", + "ext/tag.cc", + "ext/timeval.cc" ], 'conditions' : [ ['no_install=="yes"', { diff --git a/src/node/byte_buffer.cc b/src/node/ext/byte_buffer.cc index 142951475a..142951475a 100644 --- a/src/node/byte_buffer.cc +++ b/src/node/ext/byte_buffer.cc diff --git a/src/node/byte_buffer.h b/src/node/ext/byte_buffer.h index ee2b4c0d15..ee2b4c0d15 100644 --- a/src/node/byte_buffer.h +++ b/src/node/ext/byte_buffer.h diff --git a/src/node/call.cc b/src/node/ext/call.cc index 6434c2f0d5..6434c2f0d5 100644 --- a/src/node/call.cc +++ b/src/node/ext/call.cc diff --git a/src/node/call.h b/src/node/ext/call.h index 1924a1bf42..1924a1bf42 100644 --- a/src/node/call.h +++ b/src/node/ext/call.h diff --git a/src/node/channel.cc b/src/node/ext/channel.cc index 9087d6f919..9087d6f919 100644 --- a/src/node/channel.cc +++ b/src/node/ext/channel.cc diff --git a/src/node/channel.h b/src/node/ext/channel.h index 140cbf201a..140cbf201a 100644 --- a/src/node/channel.h +++ b/src/node/ext/channel.h diff --git a/src/node/completion_queue_async_worker.cc b/src/node/ext/completion_queue_async_worker.cc index 8de7db66d5..8de7db66d5 100644 --- a/src/node/completion_queue_async_worker.cc +++ b/src/node/ext/completion_queue_async_worker.cc diff --git a/src/node/completion_queue_async_worker.h b/src/node/ext/completion_queue_async_worker.h index 2c928b7024..2c928b7024 100644 --- a/src/node/completion_queue_async_worker.h +++ b/src/node/ext/completion_queue_async_worker.h diff --git a/src/node/credentials.cc b/src/node/ext/credentials.cc index f9cd2fcfe0..c8859ed941 100644 --- a/src/node/credentials.cc +++ b/src/node/ext/credentials.cc @@ -157,8 +157,7 @@ NAN_METHOD(Credentials::CreateSsl) { } NanReturnValue(WrapStruct(grpc_ssl_credentials_create( - root_certs, - key_cert_pair.private_key == NULL ? NULL : &key_cert_pair))); + root_certs, key_cert_pair.private_key == NULL ? NULL : &key_cert_pair))); } NAN_METHOD(Credentials::CreateComposite) { diff --git a/src/node/credentials.h b/src/node/ext/credentials.h index 981e5a99bc..981e5a99bc 100644 --- a/src/node/credentials.h +++ b/src/node/ext/credentials.h diff --git a/src/node/event.cc b/src/node/ext/event.cc index 2ca38b7448..2ca38b7448 100644 --- a/src/node/event.cc +++ b/src/node/ext/event.cc diff --git a/src/node/event.h b/src/node/ext/event.h index e06d8f0168..e06d8f0168 100644 --- a/src/node/event.h +++ b/src/node/ext/event.h diff --git a/src/node/node_grpc.cc b/src/node/ext/node_grpc.cc index bc1dfaf899..bc1dfaf899 100644 --- a/src/node/node_grpc.cc +++ b/src/node/ext/node_grpc.cc diff --git a/src/node/server.cc b/src/node/ext/server.cc index b102775d33..b102775d33 100644 --- a/src/node/server.cc +++ b/src/node/ext/server.cc diff --git a/src/node/server.h b/src/node/ext/server.h index d50f1fb6c5..d50f1fb6c5 100644 --- a/src/node/server.h +++ b/src/node/ext/server.h diff --git a/src/node/server_credentials.cc b/src/node/ext/server_credentials.cc index 393f3a6305..393f3a6305 100644 --- a/src/node/server_credentials.cc +++ b/src/node/ext/server_credentials.cc diff --git a/src/node/server_credentials.h b/src/node/ext/server_credentials.h index 8baae3f185..8baae3f185 100644 --- a/src/node/server_credentials.h +++ b/src/node/ext/server_credentials.h diff --git a/src/node/tag.cc b/src/node/ext/tag.cc index dc8e523e12..dc8e523e12 100644 --- a/src/node/tag.cc +++ b/src/node/ext/tag.cc diff --git a/src/node/tag.h b/src/node/ext/tag.h index bdb09252d9..bdb09252d9 100644 --- a/src/node/tag.h +++ b/src/node/ext/tag.h diff --git a/src/node/timeval.cc b/src/node/ext/timeval.cc index 687e33576b..687e33576b 100644 --- a/src/node/timeval.cc +++ b/src/node/ext/timeval.cc diff --git a/src/node/timeval.h b/src/node/ext/timeval.h index 1fb0f2c690..1fb0f2c690 100644 --- a/src/node/timeval.h +++ b/src/node/ext/timeval.h diff --git a/src/node/main.js b/src/node/index.js index 751c3525d3..0627e7f557 100644 --- a/src/node/main.js +++ b/src/node/index.js @@ -35,9 +35,9 @@ var _ = require('underscore'); var ProtoBuf = require('protobufjs'); -var surface_client = require('./surface_client.js'); +var surface_client = require('./src/surface_client.js'); -var surface_server = require('./surface_server.js'); +var surface_server = require('./src/surface_server.js'); var grpc = require('bindings')('grpc'); diff --git a/src/node/package.json b/src/node/package.json index 5f3c6fa345..028dc20555 100644 --- a/src/node/package.json +++ b/src/node/package.json @@ -13,9 +13,8 @@ "underscore.string": "^3.0.0" }, "devDependencies": { - "highland": "~2.2.0", "mocha": "~1.21.0", "minimist": "^1.1.0" }, - "main": "main.js" + "main": "index.js" } diff --git a/src/node/client.js b/src/node/src/client.js index 3a1c9eef84..3a1c9eef84 100644 --- a/src/node/client.js +++ b/src/node/src/client.js diff --git a/src/node/common.js b/src/node/src/common.js index 54247e3fa1..54247e3fa1 100644 --- a/src/node/common.js +++ b/src/node/src/common.js diff --git a/src/node/server.js b/src/node/src/server.js index 03cdbe6f98..03cdbe6f98 100644 --- a/src/node/server.js +++ b/src/node/src/server.js diff --git a/src/node/surface_client.js b/src/node/src/surface_client.js index 16c31809f4..16c31809f4 100644 --- a/src/node/surface_client.js +++ b/src/node/src/surface_client.js diff --git a/src/node/surface_server.js b/src/node/src/surface_server.js index af23ec211c..af23ec211c 100644 --- a/src/node/surface_server.js +++ b/src/node/src/surface_server.js diff --git a/src/node/test/call_test.js b/src/node/test/call_test.js index 6e52ec89bd..b37c44abaf 100644 --- a/src/node/test/call_test.js +++ b/src/node/test/call_test.js @@ -34,8 +34,6 @@ var assert = require('assert'); var grpc = require('bindings')('grpc.node'); -var channel = new grpc.Channel('localhost:7070'); - /** * Helper function to return an absolute deadline given a relative timeout in * seconds. @@ -49,6 +47,17 @@ function getDeadline(timeout_secs) { } describe('call', function() { + var channel; + var server; + before(function() { + server = new grpc.Server(); + var port = server.addHttp2Port('localhost:0'); + server.start(); + channel = new grpc.Channel('localhost:' + port); + }); + after(function() { + server.shutdown(); + }); describe('constructor', function() { it('should reject anything less than 3 arguments', function() { assert.throws(function() { diff --git a/src/node/test/client_server_test.js b/src/node/test/client_server_test.js index 99438a1659..9e1b2a79c8 100644 --- a/src/node/test/client_server_test.js +++ b/src/node/test/client_server_test.js @@ -35,10 +35,9 @@ var assert = require('assert'); var fs = require('fs'); var path = require('path'); var grpc = require('bindings')('grpc.node'); -var Server = require('../server'); -var client = require('../client'); -var common = require('../common'); -var _ = require('highland'); +var Server = require('../src/server'); +var client = require('../src/client'); +var common = require('../src/common'); var ca_path = path.join(__dirname, 'data/ca.pem'); @@ -85,38 +84,65 @@ function cancelHandler(stream) { // do nothing } +/** + * Serialize a string to a Buffer + * @param {string} value The string to serialize + * @return {Buffer} The serialized value + */ +function stringSerialize(value) { + return new Buffer(value); +} + +/** + * Deserialize a Buffer to a string + * @param {Buffer} buffer The buffer to deserialize + * @return {string} The string value of the buffer + */ +function stringDeserialize(buffer) { + return buffer.toString(); +} + describe('echo client', function() { - it('should receive echo responses', function(done) { - var server = new Server(); + var server; + var channel; + before(function() { + server = new Server(); var port_num = server.bind('0.0.0.0:0'); server.register('echo', echoHandler); + server.register('error', errorHandler); + server.register('cancellation', cancelHandler); server.start(); + channel = new grpc.Channel('localhost:' + port_num); + }); + after(function() { + server.shutdown(); + }); + it('should receive echo responses', function(done) { var messages = ['echo1', 'echo2', 'echo3', 'echo4']; - var channel = new grpc.Channel('localhost:' + port_num); var stream = client.makeRequest( channel, - 'echo'); - _(messages).map(function(val) { - return new Buffer(val); - }).pipe(stream); + 'echo', + stringSerialize, + stringDeserialize); + for (var i = 0; i < messages.length; i++) { + stream.write(messages[i]); + } + stream.end(); var index = 0; stream.on('data', function(chunk) { - assert.equal(messages[index], chunk.toString()); + assert.equal(messages[index], chunk); index += 1; }); + stream.on('status', function(status) { + assert.equal(status.code, client.status.OK); + }); stream.on('end', function() { - server.shutdown(); + assert.equal(index, messages.length); done(); }); }); it('should get an error status that the server throws', function(done) { - var server = new Server(); - var port_num = server.bind('0.0.0.0:0'); - server.register('error', errorHandler); - server.start(); - - var channel = new grpc.Channel('localhost:' + port_num); var stream = client.makeRequest( channel, 'error', @@ -129,17 +155,10 @@ describe('echo client', function() { stream.on('status', function(status) { assert.equal(status.code, grpc.status.UNIMPLEMENTED); assert.equal(status.details, 'error details'); - server.shutdown(); done(); }); }); it('should be able to cancel a call', function(done) { - var server = new Server(); - var port_num = server.bind('0.0.0.0:0'); - server.register('cancellation', cancelHandler); - server.start(); - - var channel = new grpc.Channel('localhost:' + port_num); var stream = client.makeRequest( channel, 'cancellation', @@ -149,7 +168,6 @@ describe('echo client', function() { stream.cancel(); stream.on('status', function(status) { assert.equal(status.code, grpc.status.CANCELLED); - server.shutdown(); done(); }); }); @@ -157,7 +175,9 @@ describe('echo client', function() { /* TODO(mlumish): explore options for reducing duplication between this test * and the insecure echo client test */ describe('secure echo client', function() { - it('should recieve echo responses', function(done) { + var server; + var channel; + before(function(done) { fs.readFile(ca_path, function(err, ca_data) { assert.ifError(err); fs.readFile(key_path, function(err, key_data) { @@ -169,34 +189,45 @@ describe('secure echo client', function() { key_data, pem_data); - var server = new Server({'credentials' : server_creds}); + server = new Server({'credentials' : server_creds}); var port_num = server.bind('0.0.0.0:0', true); server.register('echo', echoHandler); server.start(); - var messages = ['echo1', 'echo2', 'echo3', 'echo4']; - var channel = new grpc.Channel('localhost:' + port_num, { + channel = new grpc.Channel('localhost:' + port_num, { 'grpc.ssl_target_name_override' : 'foo.test.google.com', 'credentials' : creds }); - var stream = client.makeRequest( - channel, - 'echo'); - - _(messages).map(function(val) { - return new Buffer(val); - }).pipe(stream); - var index = 0; - stream.on('data', function(chunk) { - assert.equal(messages[index], chunk.toString()); - index += 1; - }); - stream.on('end', function() { - server.shutdown(); - done(); - }); + done(); }); }); }); }); + after(function() { + server.shutdown(); + }); + it('should recieve echo responses', function(done) { + var messages = ['echo1', 'echo2', 'echo3', 'echo4']; + var stream = client.makeRequest( + channel, + 'echo', + stringSerialize, + stringDeserialize); + for (var i = 0; i < messages.length; i++) { + stream.write(messages[i]); + } + stream.end(); + var index = 0; + stream.on('data', function(chunk) { + assert.equal(messages[index], chunk); + index += 1; + }); + stream.on('status', function(status) { + assert.equal(status.code, client.status.OK); + }); + stream.on('end', function() { + assert.equal(index, messages.length); + done(); + }); + }); }); diff --git a/src/node/test/end_to_end_test.js b/src/node/test/end_to_end_test.js index f7ccbcf5f2..f8cb660d2d 100644 --- a/src/node/test/end_to_end_test.js +++ b/src/node/test/end_to_end_test.js @@ -56,14 +56,21 @@ function multiDone(done, count) { } describe('end-to-end', function() { + var server; + var channel; + before(function() { + server = new grpc.Server(); + var port_num = server.addHttp2Port('0.0.0.0:0'); + server.start(); + channel = new grpc.Channel('localhost:' + port_num); + }); + after(function() { + server.shutdown(); + }); it('should start and end a request without error', function(complete) { - var server = new grpc.Server(); var done = multiDone(function() { complete(); - server.shutdown(); }, 2); - var port_num = server.addHttp2Port('0.0.0.0:0'); - var channel = new grpc.Channel('localhost:' + port_num); var deadline = new Date(); deadline.setSeconds(deadline.getSeconds() + 3); var status_text = 'xyz'; @@ -81,7 +88,6 @@ describe('end-to-end', function() { done(); }, 0); - server.start(); server.requestCall(function(event) { assert.strictEqual(event.type, grpc.completionType.SERVER_RPC_NEW); var server_call = event.call; @@ -109,13 +115,10 @@ describe('end-to-end', function() { it('should send and receive data without error', function(complete) { var req_text = 'client_request'; var reply_text = 'server_response'; - var server = new grpc.Server(); var done = multiDone(function() { complete(); server.shutdown(); }, 6); - var port_num = server.addHttp2Port('0.0.0.0:0'); - var channel = new grpc.Channel('localhost:' + port_num); var deadline = new Date(); deadline.setSeconds(deadline.getSeconds() + 3); var status_text = 'success'; @@ -151,8 +154,6 @@ describe('end-to-end', function() { assert.strictEqual(event.data.toString(), reply_text); done(); }); - - server.start(); server.requestCall(function(event) { assert.strictEqual(event.type, grpc.completionType.SERVER_RPC_NEW); var server_call = event.call; diff --git a/src/node/test/server_test.js b/src/node/test/server_test.js index 457d13d2f5..5fad9a5564 100644 --- a/src/node/test/server_test.js +++ b/src/node/test/server_test.js @@ -33,7 +33,7 @@ var assert = require('assert'); var grpc = require('bindings')('grpc.node'); -var Server = require('../server'); +var Server = require('../src/server'); /** * This is used for testing functions with multiple asynchronous calls that @@ -65,17 +65,22 @@ function echoHandler(stream) { } describe('echo server', function() { - it('should echo inputs as responses', function(done) { - done = multiDone(done, 4); - var server = new Server(); + var server; + var channel; + before(function() { + server = new Server(); var port_num = server.bind('[::]:0'); server.register('echo', echoHandler); server.start(); + channel = new grpc.Channel('localhost:' + port_num); + }); + it('should echo inputs as responses', function(done) { + done = multiDone(done, 4); + var req_text = 'echo test string'; var status_text = 'OK'; - var channel = new grpc.Channel('localhost:' + port_num); var deadline = new Date(); deadline.setSeconds(deadline.getSeconds() + 3); var call = new grpc.Call(channel, diff --git a/src/node/test/surface_test.js b/src/node/test/surface_test.js index 16e4869d83..85f4841d4b 100644 --- a/src/node/test/surface_test.js +++ b/src/node/test/surface_test.js @@ -33,9 +33,9 @@ var assert = require('assert'); -var surface_server = require('../surface_server.js'); +var surface_server = require('../src/surface_server.js'); -var surface_client = require('../surface_client.js'); +var surface_client = require('../src/surface_client.js'); var ProtoBuf = require('protobufjs'); diff --git a/src/ruby/bin/interop/interop_client.rb b/src/ruby/bin/interop/interop_client.rb index 0ea7f376be..86739b7b67 100755 --- a/src/ruby/bin/interop/interop_client.rb +++ b/src/ruby/bin/interop/interop_client.rb @@ -54,6 +54,8 @@ require 'test/cpp/interop/test_services' require 'test/cpp/interop/messages' require 'test/cpp/interop/empty' +require 'signet/ssl_config' + # loads the certificates used to access the test server securely. def load_test_certs this_dir = File.expand_path(File.dirname(__FILE__)) @@ -62,21 +64,49 @@ def load_test_certs files.map { |f| File.open(File.join(data_dir, f)).read } end +# loads the certificates used to access the test server securely. +def load_prod_cert + fail 'could not find a production cert' if ENV['SSL_CERT_FILE'].nil? + p "loading prod certs from #{ENV['SSL_CERT_FILE']}" + File.open(ENV['SSL_CERT_FILE']).read +end + # creates a Credentials from the test certificates. def test_creds certs = load_test_certs GRPC::Core::Credentials.new(certs[0]) end +RX_CERT = /-----BEGIN CERTIFICATE-----\n.*?-----END CERTIFICATE-----\n/m + + +# creates a Credentials from the production certificates. +def prod_creds + cert_text = load_prod_cert + GRPC::Core::Credentials.new(cert_text) +end + # creates a test stub that accesses host:port securely. -def create_stub(host, port) +def create_stub(host, port, is_secure, host_override, use_test_ca) address = "#{host}:#{port}" - stub_opts = { - :creds => test_creds, - GRPC::Core::Channel::SSL_TARGET => 'foo.test.google.com' - } - logger.info("... connecting securely to #{address}") - Grpc::Testing::TestService::Stub.new(address, **stub_opts) + if is_secure + creds = nil + if use_test_ca + creds = test_creds + else + creds = prod_creds + end + + stub_opts = { + :creds => creds, + GRPC::Core::Channel::SSL_TARGET => host_override + } + logger.info("... connecting securely to #{address}") + Grpc::Testing::TestService::Stub.new(address, **stub_opts) + else + logger.info("... connecting insecurely to #{address}") + Grpc::Testing::TestService::Stub.new(address) + end end # produces a string of null chars (\0) of length l. @@ -133,20 +163,12 @@ class NamedTests @stub = stub end - # TESTING - # PASSED - # FAIL - # ruby server: fails protobuf-ruby can't pass an empty message def empty_unary resp = @stub.empty_call(Empty.new) assert resp.is_a?(Empty), 'empty_unary: invalid response' p 'OK: empty_unary' end - # TESTING - # PASSED - # ruby server - # FAILED def large_unary req_size, wanted_response_size = 271_828, 314_159 payload = Payload.new(type: :COMPRESSABLE, body: nulls(req_size)) @@ -163,10 +185,6 @@ class NamedTests p 'OK: large_unary' end - # TESTING: - # PASSED - # ruby server - # FAILED def client_streaming msg_sizes = [27_182, 8, 1828, 45_904] wanted_aggregate_size = 74_922 @@ -180,10 +198,6 @@ class NamedTests p 'OK: client_streaming' end - # TESTING: - # PASSED - # ruby server - # FAILED def server_streaming msg_sizes = [31_415, 9, 2653, 58_979] response_spec = msg_sizes.map { |s| ResponseParameters.new(size: s) } @@ -200,10 +214,6 @@ class NamedTests p 'OK: server_streaming' end - # TESTING: - # PASSED - # ruby server - # FAILED def ping_pong msg_sizes = [[27_182, 31_415], [8, 9], [1828, 2653], [45_904, 58_979]] ppp = PingPongPlayer.new(msg_sizes) @@ -211,12 +221,23 @@ class NamedTests resps.each { |r| ppp.queue.push(r) } p 'OK: ping_pong' end + + def all + all_methods = NamedTests.instance_methods(false).map(&:to_s) + all_methods.each do |m| + next if m == 'all' || m.start_with?('assert') + p "TESTCASE: #{m}" + method(m).call + end + end end # validates the the command line options, returning them as a Hash. def parse_options options = { + 'secure' => false, 'server_host' => nil, + 'server_host_override' => nil, 'server_port' => nil, 'test_case' => nil } @@ -225,6 +246,10 @@ def parse_options opts.on('--server_host SERVER_HOST', 'server hostname') do |v| options['server_host'] = v end + opts.on('--server_host_override HOST_OVERRIDE', + 'override host via a HTTP header') do |v| + options['server_host_override'] = v + end opts.on('--server_port SERVER_PORT', 'server port') do |v| options['server_port'] = v end @@ -235,19 +260,33 @@ def parse_options " (#{test_case_list})") do |v| options['test_case'] = v end + opts.on('-s', '--use_tls', 'require a secure connection?') do |v| + options['secure'] = v + end + opts.on('-t', '--use_test_ca', + 'if secure, use the test certificate?') do |v| + options['use_test_ca'] = v + end end.parse! + _check_options(options) +end +def _check_options(opts) %w(server_host server_port test_case).each do |arg| - if options[arg].nil? + if opts[arg].nil? fail(OptionParser::MissingArgument, "please specify --#{arg}") end end - options + if opts['server_host_override'].nil? + opts['server_host_override'] = opts['server_host'] + end + opts end def main opts = parse_options - stub = create_stub(opts['server_host'], opts['server_port']) + stub = create_stub(opts['server_host'], opts['server_port'], opts['secure'], + opts['server_host_override'], opts['use_test_ca']) NamedTests.new(stub).method(opts['test_case']).call end diff --git a/src/ruby/bin/interop/interop_server.rb b/src/ruby/bin/interop/interop_server.rb index 83212823f6..cc4d260879 100755 --- a/src/ruby/bin/interop/interop_server.rb +++ b/src/ruby/bin/interop/interop_server.rb @@ -154,13 +154,17 @@ end # validates the the command line options, returning them as a Hash. def parse_options options = { - 'port' => nil + 'port' => nil, + 'secure' => false } OptionParser.new do |opts| opts.banner = 'Usage: --port port' opts.on('--port PORT', 'server port') do |v| options['port'] = v end + opts.on('-s', '--use_tls', 'require a secure connection?') do |v| + options['secure'] = v + end end.parse! if options['port'].nil? @@ -172,10 +176,15 @@ end def main opts = parse_options host = "0.0.0.0:#{opts['port']}" - s = GRPC::RpcServer.new(creds: test_server_creds) - s.add_http2_port(host, true) - logger.info("... running securely on #{host}") - + if opts['secure'] + s = GRPC::RpcServer.new(creds: test_server_creds) + s.add_http2_port(host, true) + logger.info("... running securely on #{host}") + else + s = GRPC::RpcServer.new + s.add_http2_port(host) + logger.info("... running insecurely on #{host}") + end s.handle(TestTarget) s.run end diff --git a/src/ruby/ext/grpc/rb_channel_args.c b/src/ruby/ext/grpc/rb_channel_args.c index b918e1264e..cf49259128 100644 --- a/src/ruby/ext/grpc/rb_channel_args.c +++ b/src/ruby/ext/grpc/rb_channel_args.c @@ -143,7 +143,7 @@ void grpc_rb_hash_convert_to_channel_args(VALUE src_hash, /* Make a protected call to grpc_rb_hash_convert_channel_args */ params.src_hash = src_hash; params.dst = dst; - rb_protect(grpc_rb_hash_convert_to_channel_args0, (VALUE)¶ms, &status); + rb_protect(grpc_rb_hash_convert_to_channel_args0, (VALUE) & params, &status); if (status != 0) { if (dst->args != NULL) { /* Free any allocated memory before propagating the error */ diff --git a/src/ruby/ext/grpc/rb_credentials.c b/src/ruby/ext/grpc/rb_credentials.c index 31f47f3b76..87a5d0a299 100644 --- a/src/ruby/ext/grpc/rb_credentials.c +++ b/src/ruby/ext/grpc/rb_credentials.c @@ -84,7 +84,6 @@ static void grpc_rb_credentials_mark(void *p) { } /* Allocates Credential instances. - Provides safe initial defaults for the instance fields. */ static VALUE grpc_rb_credentials_alloc(VALUE cls) { grpc_rb_credentials *wrapper = ALLOC(grpc_rb_credentials); @@ -95,7 +94,6 @@ static VALUE grpc_rb_credentials_alloc(VALUE cls) { } /* Clones Credentials instances. - Gives Credentials a consistent implementation of Ruby's object copy/dup protocol. */ static VALUE grpc_rb_credentials_init_copy(VALUE copy, VALUE orig) { @@ -124,7 +122,6 @@ static VALUE grpc_rb_credentials_init_copy(VALUE copy, VALUE orig) { /* call-seq: creds = Credentials.default() - Creates the default credential instances. */ static VALUE grpc_rb_default_credentials_create(VALUE cls) { grpc_rb_credentials *wrapper = ALLOC(grpc_rb_credentials); @@ -143,7 +140,6 @@ static VALUE grpc_rb_default_credentials_create(VALUE cls) { /* call-seq: creds = Credentials.compute_engine() - Creates the default credential instances. */ static VALUE grpc_rb_compute_engine_credentials_create(VALUE cls) { grpc_rb_credentials *wrapper = ALLOC(grpc_rb_credentials); @@ -164,7 +160,6 @@ static VALUE grpc_rb_compute_engine_credentials_create(VALUE cls) { creds1 = ... creds2 = ... creds3 = creds1.add(creds2) - Creates the default credential instances. */ static VALUE grpc_rb_composite_credentials_create(VALUE self, VALUE other) { grpc_rb_credentials *self_wrapper = NULL; @@ -202,11 +197,9 @@ static ID id_pem_cert_chain; ... creds2 = Credentials.new(pem_root_certs, pem_private_key, pem_cert_chain) - pem_root_certs: (required) PEM encoding of the server root certificate pem_private_key: (optional) PEM encoding of the client's private key pem_cert_chain: (optional) PEM encoding of the client's cert chain - Initializes Credential instances. */ static VALUE grpc_rb_credentials_init(int argc, VALUE *argv, VALUE self) { VALUE pem_root_certs = Qnil; @@ -214,6 +207,8 @@ static VALUE grpc_rb_credentials_init(int argc, VALUE *argv, VALUE self) { VALUE pem_cert_chain = Qnil; grpc_rb_credentials *wrapper = NULL; grpc_credentials *creds = NULL; + grpc_ssl_pem_key_cert_pair key_cert_pair; + MEMZERO(&key_cert_pair, grpc_ssl_pem_key_cert_pair, 1); /* TODO: Remove mandatory arg when we support default roots. */ /* "12" == 1 mandatory arg, 2 (credentials) is optional */ rb_scan_args(argc, argv, "12", &pem_root_certs, &pem_private_key, @@ -228,8 +223,8 @@ static VALUE grpc_rb_credentials_init(int argc, VALUE *argv, VALUE self) { if (pem_private_key == Qnil && pem_cert_chain == Qnil) { creds = grpc_ssl_credentials_create(RSTRING_PTR(pem_root_certs), NULL); } else { - grpc_ssl_pem_key_cert_pair key_cert_pair = {RSTRING_PTR(pem_private_key), - RSTRING_PTR(pem_cert_chain)}; + key_cert_pair.private_key = RSTRING_PTR(pem_private_key); + key_cert_pair.cert_chain = RSTRING_PTR(pem_cert_chain); creds = grpc_ssl_credentials_create( RSTRING_PTR(pem_root_certs), &key_cert_pair); } diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c index ef2a9f107b..436d006760 100644 --- a/src/ruby/ext/grpc/rb_server.c +++ b/src/ruby/ext/grpc/rb_server.c @@ -223,7 +223,7 @@ static VALUE grpc_rb_server_add_http2_port(int argc, VALUE *argv, VALUE self) { VALUE port = Qnil; VALUE is_secure = Qnil; grpc_rb_server *s = NULL; - int added_ok = 0; + int recvd_port = 0; /* "11" == 1 mandatory args, 1 (is_secure) is optional */ rb_scan_args(argc, argv, "11", &port, &is_secure); @@ -233,22 +233,22 @@ static VALUE grpc_rb_server_add_http2_port(int argc, VALUE *argv, VALUE self) { rb_raise(rb_eRuntimeError, "closed!"); return Qnil; } else if (is_secure == Qnil || TYPE(is_secure) != T_TRUE) { - added_ok = grpc_server_add_http2_port(s->wrapped, StringValueCStr(port)); - if (added_ok == 0) { + recvd_port = grpc_server_add_http2_port(s->wrapped, StringValueCStr(port)); + if (recvd_port == 0) { rb_raise(rb_eRuntimeError, "could not add port %s to server, not sure why", StringValueCStr(port)); } } else if (TYPE(is_secure) != T_FALSE) { - added_ok = + recvd_port = grpc_server_add_secure_http2_port(s->wrapped, StringValueCStr(port)); - if (added_ok == 0) { + if (recvd_port == 0) { rb_raise(rb_eRuntimeError, "could not add secure port %s to server, not sure why", StringValueCStr(port)); } } - return Qnil; + return INT2NUM(recvd_port); } void Init_google_rpc_server() { diff --git a/src/ruby/grpc.gemspec b/src/ruby/grpc.gemspec index 450362f5a8..ffd084dc91 100755 --- a/src/ruby/grpc.gemspec +++ b/src/ruby/grpc.gemspec @@ -22,6 +22,7 @@ Gem::Specification.new do |s| s.add_dependency 'xray' s.add_dependency 'logging', '~> 1.8' s.add_dependency 'google-protobuf', '~> 3.0.0alpha.1.1' + s.add_dependency 'signet', '~> 0.5.1' s.add_dependency 'minitest', '~> 5.4' # reqd for interop tests s.add_development_dependency 'bundler', '~> 1.7' diff --git a/src/ruby/lib/grpc/generic/active_call.rb b/src/ruby/lib/grpc/generic/active_call.rb index 1cdc168bfe..6c2b6e91c2 100644 --- a/src/ruby/lib/grpc/generic/active_call.rb +++ b/src/ruby/lib/grpc/generic/active_call.rb @@ -67,7 +67,7 @@ module Google fail(ArgumentError, 'not a CompletionQueue') end call.add_metadata(kw) if kw.length > 0 - invoke_accepted, client_metadata_read = Object.new, Object.new + client_metadata_read = Object.new finished_tag = Object.new call.invoke(q, client_metadata_read, finished_tag) [finished_tag, client_metadata_read] diff --git a/src/ruby/spec/call_spec.rb b/src/ruby/spec/call_spec.rb index 9a510df1f3..c793284488 100644 --- a/src/ruby/spec/call_spec.rb +++ b/src/ruby/spec/call_spec.rb @@ -28,7 +28,6 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. require 'grpc' -require 'port_picker' include GRPC::Core::StatusCodes @@ -71,16 +70,8 @@ describe GRPC::Core::Call do before(:each) do @tag = Object.new @client_queue = GRPC::Core::CompletionQueue.new - @server_queue = GRPC::Core::CompletionQueue.new - port = find_unused_tcp_port - host = "localhost:#{port}" - @server = GRPC::Core::Server.new(@server_queue, nil) - @server.add_http2_port(host) - @ch = GRPC::Core::Channel.new(host, nil) - end - - after(:each) do - @server.close + fake_host = 'localhost:10101' + @ch = GRPC::Core::Channel.new(fake_host, nil) end describe '#start_read' do @@ -122,19 +113,6 @@ describe GRPC::Core::Call do end end - describe '#start_write' do - it 'should cause the WRITE_ACCEPTED event' do - call = make_test_call - call.invoke(@client_queue, @tag, @tag) - expect(call.start_write(GRPC::Core::ByteBuffer.new('test_start_write'), - @tag)).to be_nil - ev = @client_queue.next(deadline) - expect(ev.call).to be_a(GRPC::Core::Call) - expect(ev.type).to be(GRPC::Core::CompletionType::WRITE_ACCEPTED) - expect(ev.tag).to be(@tag) - end - end - describe '#status' do it 'can save the status and read it back' do call = make_test_call diff --git a/src/ruby/spec/channel_spec.rb b/src/ruby/spec/channel_spec.rb index 820dbd39e9..189d1c67ab 100644 --- a/src/ruby/spec/channel_spec.rb +++ b/src/ruby/spec/channel_spec.rb @@ -28,7 +28,8 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. require 'grpc' -require 'port_picker' + +FAKE_HOST='localhost:0' def load_test_certs test_root = File.join(File.dirname(__FILE__), 'testdata') @@ -114,8 +115,7 @@ describe GRPC::Core::Channel do describe '#create_call' do it 'creates a call OK' do - port = find_unused_tcp_port - host = "localhost:#{port}" + host = FAKE_HOST ch = GRPC::Core::Channel.new(host, nil) deadline = Time.now + 5 @@ -127,8 +127,7 @@ describe GRPC::Core::Channel do end it 'raises an error if called on a closed channel' do - port = find_unused_tcp_port - host = "localhost:#{port}" + host = FAKE_HOST ch = GRPC::Core::Channel.new(host, nil) ch.close @@ -142,16 +141,14 @@ describe GRPC::Core::Channel do describe '#destroy' do it 'destroys a channel ok' do - port = find_unused_tcp_port - host = "localhost:#{port}" + host = FAKE_HOST ch = GRPC::Core::Channel.new(host, nil) blk = proc { ch.destroy } expect(&blk).to_not raise_error end it 'can be called more than once without error' do - port = find_unused_tcp_port - host = "localhost:#{port}" + host = FAKE_HOST ch = GRPC::Core::Channel.new(host, nil) blk = proc { ch.destroy } blk.call @@ -167,16 +164,14 @@ describe GRPC::Core::Channel do describe '#close' do it 'closes a channel ok' do - port = find_unused_tcp_port - host = "localhost:#{port}" + host = FAKE_HOST ch = GRPC::Core::Channel.new(host, nil) blk = proc { ch.close } expect(&blk).to_not raise_error end it 'can be called more than once without error' do - port = find_unused_tcp_port - host = "localhost:#{port}" + host = FAKE_HOST ch = GRPC::Core::Channel.new(host, nil) blk = proc { ch.close } blk.call diff --git a/src/ruby/spec/client_server_spec.rb b/src/ruby/spec/client_server_spec.rb index b2afb0581e..96b8ef4300 100644 --- a/src/ruby/spec/client_server_spec.rb +++ b/src/ruby/spec/client_server_spec.rb @@ -28,7 +28,6 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. require 'grpc' -require 'port_picker' require 'spec_helper' include GRPC::Core::CompletionType @@ -44,12 +43,13 @@ shared_context 'setup: tags' do before(:example) do @server_finished_tag = Object.new @client_finished_tag = Object.new + @client_metadata_tag = Object.new @server_tag = Object.new @tag = Object.new end def deadline - Time.now + 0.05 + Time.now + 2 end def expect_next_event_on(queue, type, tag) @@ -63,27 +63,30 @@ shared_context 'setup: tags' do ev end - def server_receives_and_responds_with(reply_text) - reply = ByteBuffer.new(reply_text) + def server_allows_client_to_proceed @server.request_call(@server_tag) - ev = @server_queue.pluck(@server_tag, TimeConsts::INFINITE_FUTURE) + ev = @server_queue.pluck(@server_tag, deadline) expect(ev).not_to be_nil expect(ev.type).to be(SERVER_RPC_NEW) - ev.call.server_accept(@server_queue, @server_finished_tag) - ev.call.server_end_initial_metadata - ev.call.start_read(@server_tag) + server_call = ev.call + server_call.server_accept(@server_queue, @server_finished_tag) + server_call.server_end_initial_metadata + server_call + end + + def server_responds_with(server_call, reply_text) + reply = ByteBuffer.new(reply_text) + server_call.start_read(@server_tag) ev = @server_queue.pluck(@server_tag, TimeConsts::INFINITE_FUTURE) expect(ev.type).to be(READ) - ev.call.start_write(reply, @server_tag) + server_call.start_write(reply, @server_tag) ev = @server_queue.pluck(@server_tag, TimeConsts::INFINITE_FUTURE) expect(ev).not_to be_nil expect(ev.type).to be(WRITE_ACCEPTED) - ev.call end def client_sends(call, sent = 'a message') req = ByteBuffer.new(sent) - call.invoke(@client_queue, @tag, @client_finished_tag) call.start_write(req, @tag) ev = @client_queue.pluck(@tag, TimeConsts::INFINITE_FUTURE) expect(ev).not_to be_nil @@ -102,16 +105,20 @@ shared_examples 'basic GRPC message delivery is OK' do it 'servers receive requests from clients and start responding' do reply = ByteBuffer.new('the server payload') call = new_client_call - msg = client_sends(call) + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) # check the server rpc new was received - @server.request_call(@server_tag) - ev = expect_next_event_on(@server_queue, SERVER_RPC_NEW, @server_tag) + # @server.request_call(@server_tag) + # ev = expect_next_event_on(@server_queue, SERVER_RPC_NEW, @server_tag) # accept the call - server_call = ev.call - server_call.server_accept(@server_queue, @server_finished_tag) - server_call.server_end_initial_metadata + # server_call = ev.call + # server_call.server_accept(@server_queue, @server_finished_tag) + # server_call.server_end_initial_metadata + server_call = server_allows_client_to_proceed + + # client sends a message + msg = client_sends(call) # confirm the server can read the inbound message server_call.start_read(@server_tag) @@ -125,18 +132,19 @@ shared_examples 'basic GRPC message delivery is OK' do it 'responses written by servers are received by the client' do call = new_client_call + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) + server_call = server_allows_client_to_proceed client_sends(call) - server_receives_and_responds_with('server_response') + server_responds_with(server_call, 'server_response') call.start_read(@tag) - expect_next_event_on(@client_queue, CLIENT_METADATA_READ, @tag) ev = expect_next_event_on(@client_queue, READ, @tag) expect(ev.result.to_s).to eq('server_response') end it 'servers can ignore a client write and send a status' do call = new_client_call - client_sends(call) + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) # check the server rpc new was received @server.request_call(@server_tag) @@ -150,9 +158,13 @@ shared_examples 'basic GRPC message delivery is OK' do server_call.start_write_status(StatusCodes::NOT_FOUND, 'not found', @server_tag) + # Client sends some data + client_sends(call) + # client gets an empty response for the read, preceeded by some metadata. call.start_read(@tag) - expect_next_event_on(@client_queue, CLIENT_METADATA_READ, @tag) + expect_next_event_on(@client_queue, CLIENT_METADATA_READ, + @client_metadata_tag) ev = expect_next_event_on(@client_queue, READ, @tag) expect(ev.tag).to be(@tag) expect(ev.result.to_s).to eq('') @@ -166,13 +178,14 @@ shared_examples 'basic GRPC message delivery is OK' do it 'completes calls by sending status to client and server' do call = new_client_call + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) + server_call = server_allows_client_to_proceed client_sends(call) - server_call = server_receives_and_responds_with('server_response') + server_responds_with(server_call, 'server_response') server_call.start_write_status(10_101, 'status code is 10101', @server_tag) # first the client says writes are done call.start_read(@tag) - expect_next_event_on(@client_queue, CLIENT_METADATA_READ, @tag) expect_next_event_on(@client_queue, READ, @tag) call.writes_done(@tag) @@ -215,22 +228,13 @@ shared_examples 'GRPC metadata delivery works OK' do end end - it 'sends an empty hash when no metadata is added' do - call = new_client_call - client_sends(call) - - # Server gets a response - @server.request_call(@server_tag) - expect_next_event_on(@server_queue, SERVER_RPC_NEW, @server_tag) - end - it 'sends all the metadata pairs when keys and values are valid' do @valid_metadata.each do |md| call = new_client_call call.add_metadata(md) # Client begins a call OK - call.invoke(@client_queue, @tag, @client_finished_tag) + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) # ... server has all metadata available even though the client did not # send a write @@ -262,7 +266,7 @@ shared_examples 'GRPC metadata delivery works OK' do it 'raises an exception if a metadata key is invalid' do @bad_keys.each do |md| call = new_client_call - client_sends(call) + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) # server gets the invocation @server.request_call(@server_tag) @@ -273,7 +277,7 @@ shared_examples 'GRPC metadata delivery works OK' do it 'sends a hash that contains the status when no metadata is added' do call = new_client_call - client_sends(call) + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) # server gets the invocation @server.request_call(@server_tag) @@ -284,21 +288,17 @@ shared_examples 'GRPC metadata delivery works OK' do server_call.server_accept(@server_queue, @server_finished_tag) server_call.server_end_initial_metadata - # ... these server sends some data, allowing the metadata read - server_call.start_write(ByteBuffer.new('reply with metadata'), - @server_tag) - expect_next_event_on(@server_queue, WRITE_ACCEPTED, @server_tag) - # there is the HTTP status metadata, though there should not be any # TODO: update this with the bug number to be resolved - ev = expect_next_event_on(@client_queue, CLIENT_METADATA_READ, @tag) + ev = expect_next_event_on(@client_queue, CLIENT_METADATA_READ, + @client_metadata_tag) expect(ev.result).to eq(':status' => '200') end it 'sends all the pairs and status:200 when keys and values are valid' do @valid_metadata.each do |md| call = new_client_call - client_sends(call) + call.invoke(@client_queue, @client_metadata_tag, @client_finished_tag) # server gets the invocation @server.request_call(@server_tag) @@ -311,7 +311,8 @@ shared_examples 'GRPC metadata delivery works OK' do server_call.server_end_initial_metadata # Now the client can read the metadata - ev = expect_next_event_on(@client_queue, CLIENT_METADATA_READ, @tag) + ev = expect_next_event_on(@client_queue, CLIENT_METADATA_READ, + @client_metadata_tag) replace_symbols = Hash[md.each_pair.collect { |x, y| [x.to_s, y] }] replace_symbols[':status'] = '200' expect(ev.result).to eq(replace_symbols) @@ -322,17 +323,17 @@ end describe 'the http client/server' do before(:example) do - port = find_unused_tcp_port - host = "localhost:#{port}" + server_host = '0.0.0.0:0' @client_queue = GRPC::Core::CompletionQueue.new @server_queue = GRPC::Core::CompletionQueue.new @server = GRPC::Core::Server.new(@server_queue, nil) - @server.add_http2_port(host) + server_port = @server.add_http2_port(server_host) @server.start - @ch = Channel.new(host, nil) + @ch = Channel.new("0.0.0.0:#{server_port}", nil) end after(:example) do + @ch.close @server.close end @@ -346,16 +347,15 @@ end describe 'the secure http client/server' do before(:example) do certs = load_test_certs - port = find_unused_tcp_port - host = "localhost:#{port}" + server_host = 'localhost:0' @client_queue = GRPC::Core::CompletionQueue.new @server_queue = GRPC::Core::CompletionQueue.new server_creds = GRPC::Core::ServerCredentials.new(nil, certs[1], certs[2]) @server = GRPC::Core::Server.new(@server_queue, nil, server_creds) - @server.add_http2_port(host, true) + server_port = @server.add_http2_port(server_host, true) @server.start args = { Channel::SSL_TARGET => 'foo.test.google.com' } - @ch = Channel.new(host, args, + @ch = Channel.new("0.0.0.0:#{server_port}", args, GRPC::Core::Credentials.new(certs[0], nil, nil)) end diff --git a/src/ruby/spec/generic/active_call_spec.rb b/src/ruby/spec/generic/active_call_spec.rb index 443ba3d192..e81b2168b0 100644 --- a/src/ruby/spec/generic/active_call_spec.rb +++ b/src/ruby/spec/generic/active_call_spec.rb @@ -28,7 +28,6 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. require 'grpc' -require_relative '../port_picker' include GRPC::Core::StatusCodes @@ -45,12 +44,11 @@ describe GRPC::ActiveCall do @client_queue = GRPC::Core::CompletionQueue.new @server_queue = GRPC::Core::CompletionQueue.new - port = find_unused_tcp_port - host = "localhost:#{port}" + host = '0.0.0.0:0' @server = GRPC::Core::Server.new(@server_queue, nil) - @server.add_http2_port(host) + server_port = @server.add_http2_port(host) @server.start - @ch = GRPC::Core::Channel.new(host, nil) + @ch = GRPC::Core::Channel.new("localhost:#{server_port}", nil) end after(:each) do @@ -206,7 +204,7 @@ describe GRPC::ActiveCall do it 'get a nil msg before a status when an OK status is sent' do call = make_test_call done_tag, meta_tag = ActiveCall.client_invoke(call, @client_queue, - deadline) + deadline) client_call = ActiveCall.new(call, @client_queue, @pass_through, @pass_through, deadline, finished_tag: done_tag, diff --git a/src/ruby/spec/generic/client_stub_spec.rb b/src/ruby/spec/generic/client_stub_spec.rb index 8ebe48bc4c..f1500fbd44 100644 --- a/src/ruby/spec/generic/client_stub_spec.rb +++ b/src/ruby/spec/generic/client_stub_spec.rb @@ -29,9 +29,9 @@ require 'grpc' require 'xray/thread_dump_signal_handler' -require_relative '../port_picker' NOOP = proc { |x| x } +FAKE_HOST = 'localhost:0' def wakey_thread(&blk) awake_mutex, awake_cond = Mutex.new, ConditionVariable.new @@ -67,7 +67,7 @@ describe 'ClientStub' do describe '#new' do it 'can be created from a host and args' do - host = new_test_host + host = FAKE_HOST opts = { a_channel_arg: 'an_arg' } blk = proc do GRPC::ClientStub.new(host, @cq, **opts) @@ -76,7 +76,7 @@ describe 'ClientStub' do end it 'can be created with a default deadline' do - host = new_test_host + host = FAKE_HOST opts = { a_channel_arg: 'an_arg', deadline: 5 } blk = proc do GRPC::ClientStub.new(host, @cq, **opts) @@ -85,7 +85,7 @@ describe 'ClientStub' do end it 'can be created with an channel override' do - host = new_test_host + host = FAKE_HOST opts = { a_channel_arg: 'an_arg', channel_override: @ch } blk = proc do GRPC::ClientStub.new(host, @cq, **opts) @@ -94,7 +94,7 @@ describe 'ClientStub' do end it 'cannot be created with a bad channel override' do - host = new_test_host + host = FAKE_HOST blk = proc do opts = { a_channel_arg: 'an_arg', channel_override: Object.new } GRPC::ClientStub.new(host, @cq, **opts) @@ -103,7 +103,7 @@ describe 'ClientStub' do end it 'cannot be created with bad credentials' do - host = new_test_host + host = FAKE_HOST blk = proc do opts = { a_channel_arg: 'an_arg', creds: Object.new } GRPC::ClientStub.new(host, @cq, **opts) @@ -113,7 +113,7 @@ describe 'ClientStub' do it 'can be created with test test credentials' do certs = load_test_certs - host = new_test_host + host = FAKE_HOST blk = proc do opts = { GRPC::Core::Channel::SSL_TARGET => 'foo.test.google.com', @@ -133,16 +133,17 @@ describe 'ClientStub' do shared_examples 'request response' do it 'should send a request to/receive a reply from a server' do - host = new_test_host - th = run_request_response(host, @sent_msg, @resp, @pass) - stub = GRPC::ClientStub.new(host, @cq) + server_port = create_test_server + th = run_request_response(@sent_msg, @resp, @pass) + stub = GRPC::ClientStub.new("localhost:#{server_port}", @cq) expect(get_response(stub)).to eq(@resp) th.join end it 'should send metadata to the server ok' do - host = new_test_host - th = run_request_response(host, @sent_msg, @resp, @pass, + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_request_response(@sent_msg, @resp, @pass, k1: 'v1', k2: 'v2') stub = GRPC::ClientStub.new(host, @cq) expect(get_response(stub)).to eq(@resp) @@ -150,8 +151,9 @@ describe 'ClientStub' do end it 'should update the sent metadata with a provided metadata updater' do - host = new_test_host - th = run_request_response(host, @sent_msg, @resp, @pass, + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_request_response(@sent_msg, @resp, @pass, k1: 'updated-v1', k2: 'v2') update_md = proc do |md| md[:k1] = 'updated-v1' @@ -163,8 +165,9 @@ describe 'ClientStub' do end it 'should send a request when configured using an override channel' do - alt_host = new_test_host - th = run_request_response(alt_host, @sent_msg, @resp, @pass) + server_port = create_test_server + alt_host = "localhost:#{server_port}" + th = run_request_response(@sent_msg, @resp, @pass) ch = GRPC::Core::Channel.new(alt_host, nil) stub = GRPC::ClientStub.new('ignored-host', @cq, channel_override: ch) expect(get_response(stub)).to eq(@resp) @@ -172,8 +175,9 @@ describe 'ClientStub' do end it 'should raise an error if the status is not OK' do - host = new_test_host - th = run_request_response(host, @sent_msg, @resp, @fail) + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_request_response(@sent_msg, @resp, @fail) stub = GRPC::ClientStub.new(host, @cq) blk = proc { get_response(stub) } expect(&blk).to raise_error(GRPC::BadStatus) @@ -210,16 +214,18 @@ describe 'ClientStub' do end it 'should send requests to/receive a reply from a server' do - host = new_test_host - th = run_client_streamer(host, @sent_msgs, @resp, @pass) + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_client_streamer(@sent_msgs, @resp, @pass) stub = GRPC::ClientStub.new(host, @cq) expect(get_response(stub)).to eq(@resp) th.join end it 'should send metadata to the server ok' do - host = new_test_host - th = run_client_streamer(host, @sent_msgs, @resp, @pass, + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_client_streamer(@sent_msgs, @resp, @pass, k1: 'v1', k2: 'v2') stub = GRPC::ClientStub.new(host, @cq) expect(get_response(stub)).to eq(@resp) @@ -227,8 +233,9 @@ describe 'ClientStub' do end it 'should update the sent metadata with a provided metadata updater' do - host = new_test_host - th = run_client_streamer(host, @sent_msgs, @resp, @pass, + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_client_streamer(@sent_msgs, @resp, @pass, k1: 'updated-v1', k2: 'v2') update_md = proc do |md| md[:k1] = 'updated-v1' @@ -240,8 +247,9 @@ describe 'ClientStub' do end it 'should raise an error if the status is not ok' do - host = new_test_host - th = run_client_streamer(host, @sent_msgs, @resp, @fail) + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_client_streamer(@sent_msgs, @resp, @fail) stub = GRPC::ClientStub.new(host, @cq) blk = proc { get_response(stub) } expect(&blk).to raise_error(GRPC::BadStatus) @@ -278,16 +286,18 @@ describe 'ClientStub' do end it 'should send a request to/receive replies from a server' do - host = new_test_host - th = run_server_streamer(host, @sent_msg, @replys, @pass) + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_server_streamer(@sent_msg, @replys, @pass) stub = GRPC::ClientStub.new(host, @cq) expect(get_responses(stub).collect { |r| r }).to eq(@replys) th.join end it 'should raise an error if the status is not ok' do - host = new_test_host - th = run_server_streamer(host, @sent_msg, @replys, @fail) + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_server_streamer(@sent_msg, @replys, @fail) stub = GRPC::ClientStub.new(host, @cq) e = get_responses(stub) expect { e.collect { |r| r } }.to raise_error(GRPC::BadStatus) @@ -295,8 +305,9 @@ describe 'ClientStub' do end it 'should send metadata to the server ok' do - host = new_test_host - th = run_server_streamer(host, @sent_msg, @replys, @fail, + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_server_streamer(@sent_msg, @replys, @fail, k1: 'v1', k2: 'v2') stub = GRPC::ClientStub.new(host, @cq) e = get_responses(stub) @@ -305,8 +316,9 @@ describe 'ClientStub' do end it 'should update the sent metadata with a provided metadata updater' do - host = new_test_host - th = run_server_streamer(host, @sent_msg, @replys, @pass, + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_server_streamer(@sent_msg, @replys, @pass, k1: 'updated-v1', k2: 'v2') update_md = proc do |md| md[:k1] = 'updated-v1' @@ -352,8 +364,9 @@ describe 'ClientStub' do end it 'supports sending all the requests first', bidi: true do - host = new_test_host - th = run_bidi_streamer_handle_inputs_first(host, @sent_msgs, @replys, + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_bidi_streamer_handle_inputs_first(@sent_msgs, @replys, @pass) stub = GRPC::ClientStub.new(host, @cq) e = get_responses(stub) @@ -362,8 +375,9 @@ describe 'ClientStub' do end it 'supports client-initiated ping pong', bidi: true do - host = new_test_host - th = run_bidi_streamer_echo_ping_pong(host, @sent_msgs, @pass, true) + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, true) stub = GRPC::ClientStub.new(host, @cq) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@sent_msgs) @@ -377,8 +391,9 @@ describe 'ClientStub' do # they receive a message from the client. Without receiving all the # metadata, the server does not accept the call, so this test hangs. xit 'supports a server-initiated ping pong', bidi: true do - host = new_test_host - th = run_bidi_streamer_echo_ping_pong(host, @sent_msgs, @pass, false) + server_port = create_test_server + host = "localhost:#{server_port}" + th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, false) stub = GRPC::ClientStub.new(host, @cq) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@sent_msgs) @@ -410,10 +425,10 @@ describe 'ClientStub' do end end - def run_server_streamer(hostname, expected_input, replys, status, **kw) + def run_server_streamer(expected_input, replys, status, **kw) wanted_metadata = kw.clone wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(hostname, mtx, cnd) + c = expect_server_to_be_invoked(mtx, cnd) wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) end @@ -423,20 +438,19 @@ describe 'ClientStub' do end end - def run_bidi_streamer_handle_inputs_first(hostname, expected_inputs, replys, + def run_bidi_streamer_handle_inputs_first(expected_inputs, replys, status) wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(hostname, mtx, cnd) + c = expect_server_to_be_invoked(mtx, cnd) expected_inputs.each { |i| expect(c.remote_read).to eq(i) } replys.each { |r| c.remote_send(r) } c.send_status(status, status == @pass ? 'OK' : 'NOK', true) end end - def run_bidi_streamer_echo_ping_pong(hostname, expected_inputs, status, - client_starts) + def run_bidi_streamer_echo_ping_pong(expected_inputs, status, client_starts) wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(hostname, mtx, cnd) + c = expect_server_to_be_invoked(mtx, cnd) expected_inputs.each do |i| if client_starts expect(c.remote_read).to eq(i) @@ -450,10 +464,10 @@ describe 'ClientStub' do end end - def run_client_streamer(hostname, expected_inputs, resp, status, **kw) + def run_client_streamer(expected_inputs, resp, status, **kw) wanted_metadata = kw.clone wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(hostname, mtx, cnd) + c = expect_server_to_be_invoked(mtx, cnd) expected_inputs.each { |i| expect(c.remote_read).to eq(i) } wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) @@ -463,10 +477,10 @@ describe 'ClientStub' do end end - def run_request_response(hostname, expected_input, resp, status, **kw) + def run_request_response(expected_input, resp, status, **kw) wanted_metadata = kw.clone wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(hostname, mtx, cnd) + c = expect_server_to_be_invoked(mtx, cnd) expect(c.remote_read).to eq(expected_input) wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) @@ -476,32 +490,30 @@ describe 'ClientStub' do end end - def start_test_server(hostname, awake_mutex, awake_cond) - server_queue = GRPC::Core::CompletionQueue.new - @server = GRPC::Core::Server.new(server_queue, nil) - @server.add_http2_port(hostname) + def create_test_server + @server_queue = GRPC::Core::CompletionQueue.new + @server = GRPC::Core::Server.new(@server_queue, nil) + @server.add_http2_port('0.0.0.0:0') + end + + def start_test_server(awake_mutex, awake_cond) @server.start @server_tag = Object.new @server.request_call(@server_tag) awake_mutex.synchronize { awake_cond.signal } - server_queue end - def expect_server_to_be_invoked(hostname, awake_mutex, awake_cond) - server_queue = start_test_server(hostname, awake_mutex, awake_cond) - ev = server_queue.pluck(@server_tag, INFINITE_FUTURE) + def expect_server_to_be_invoked(awake_mutex, awake_cond) + start_test_server(awake_mutex, awake_cond) + ev = @server_queue.pluck(@server_tag, INFINITE_FUTURE) fail OutOfTime if ev.nil? server_call = ev.call server_call.metadata = ev.result.metadata finished_tag = Object.new - server_call.server_accept(server_queue, finished_tag) + server_call.server_accept(@server_queue, finished_tag) server_call.server_end_initial_metadata - GRPC::ActiveCall.new(server_call, server_queue, NOOP, NOOP, INFINITE_FUTURE, + GRPC::ActiveCall.new(server_call, @server_queue, NOOP, NOOP, + INFINITE_FUTURE, finished_tag: finished_tag) end - - def new_test_host - port = find_unused_tcp_port - "localhost:#{port}" - end end diff --git a/src/ruby/spec/generic/rpc_server_spec.rb b/src/ruby/spec/generic/rpc_server_spec.rb index cd4888a3b4..e083bc1e9d 100644 --- a/src/ruby/spec/generic/rpc_server_spec.rb +++ b/src/ruby/spec/generic/rpc_server_spec.rb @@ -29,7 +29,6 @@ require 'grpc' require 'xray/thread_dump_signal_handler' -require_relative '../port_picker' def load_test_certs test_root = File.join(File.dirname(File.dirname(__FILE__)), 'testdata') @@ -104,10 +103,10 @@ describe GRPC::RpcServer do @noop = proc { |x| x } @server_queue = GRPC::Core::CompletionQueue.new - port = find_unused_tcp_port - @host = "localhost:#{port}" + server_host = '0.0.0.0:0' @server = GRPC::Core::Server.new(@server_queue, nil) - @server.add_http2_port(@host) + server_port = @server.add_http2_port(server_host) + @host = "localhost:#{server_port}" @ch = GRPC::Core::Channel.new(@host, nil) end diff --git a/src/ruby/spec/port_picker.rb b/src/ruby/spec/port_picker.rb deleted file mode 100644 index 98ffbacc1b..0000000000 --- a/src/ruby/spec/port_picker.rb +++ /dev/null @@ -1,45 +0,0 @@ -# Copyright 2014, Google Inc. -# All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions are -# met: -# -# * Redistributions of source code must retain the above copyright -# notice, this list of conditions and the following disclaimer. -# * Redistributions in binary form must reproduce the above -# copyright notice, this list of conditions and the following disclaimer -# in the documentation and/or other materials provided with the -# distribution. -# * Neither the name of Google Inc. nor the names of its -# contributors may be used to endorse or promote products derived from -# this software without specific prior written permission. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -require 'socket' - -# @param [Fixnum] the minimum port number to accept -# @param [Fixnum] the maximum port number to accept -# @return [Fixnum ]a free tcp port -def find_unused_tcp_port(min = 32_768, max = 60_000) - # Allow the system to assign a port, by specifying 0. - # Loop until a port is assigned in the required range - loop do - socket = Socket.new(:INET, :STREAM, 0) - socket.bind(Addrinfo.tcp('127.0.0.1', 0)) - p = socket.local_address.ip_port - socket.close - return p if p > min && p < max - end -end diff --git a/src/ruby/spec/server_spec.rb b/src/ruby/spec/server_spec.rb index 6e5bb523de..1550ba6566 100644 --- a/src/ruby/spec/server_spec.rb +++ b/src/ruby/spec/server_spec.rb @@ -28,7 +28,6 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. require 'grpc' -require 'port_picker' def load_test_certs test_root = File.join(File.dirname(__FILE__), 'testdata') @@ -205,10 +204,8 @@ describe Server do end def start_a_server - port = find_unused_tcp_port - host = "localhost:#{port}" s = Server.new(@cq, nil) - s.add_http2_port(host) + s.add_http2_port('0.0.0.0:0') s.start s end |