diff options
Diffstat (limited to 'src')
47 files changed, 3276 insertions, 117 deletions
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c index 04a7852323..f1480bb1ae 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c @@ -19,7 +19,10 @@ #include <grpc/support/port_platform.h> #if GRPC_ARES == 1 && !defined(GRPC_UV) +#include <limits.h> +#include <stdio.h> #include <string.h> +#include <unistd.h> #include <grpc/support/alloc.h> #include <grpc/support/host_port.h> @@ -31,11 +34,14 @@ #include "src/core/ext/filters/client_channel/resolver_registry.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/iomgr/combiner.h" +#include "src/core/lib/iomgr/gethostname.h" #include "src/core/lib/iomgr/resolve_address.h" #include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/json/json.h" #include "src/core/lib/support/backoff.h" #include "src/core/lib/support/env.h" #include "src/core/lib/support/string.h" +#include "src/core/lib/transport/service_config.h" #define GRPC_DNS_MIN_CONNECT_TIMEOUT_SECONDS 1 #define GRPC_DNS_INITIAL_CONNECT_BACKOFF_SECONDS 1 @@ -54,6 +60,8 @@ typedef struct { char *default_port; /** channel args. */ grpc_channel_args *channel_args; + /** whether to request the service config */ + bool request_service_config; /** pollset_set to drive the name resolution process */ grpc_pollset_set *interested_parties; @@ -85,6 +93,8 @@ typedef struct { /** currently resolving addresses */ grpc_lb_addresses *lb_addresses; + /** currently resolving service config */ + char *service_config_json; } ares_dns_resolver; static void dns_ares_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *r); @@ -144,6 +154,77 @@ static void dns_ares_on_retry_timer_locked(grpc_exec_ctx *exec_ctx, void *arg, GRPC_RESOLVER_UNREF(exec_ctx, &r->base, "retry-timer"); } +static bool value_in_json_array(grpc_json *array, const char *value) { + for (grpc_json *entry = array->child; entry != NULL; entry = entry->next) { + if (entry->type == GRPC_JSON_STRING && strcmp(entry->value, value) == 0) { + return true; + } + } + return false; +} + +static char *choose_service_config(char *service_config_choice_json) { + grpc_json *choices_json = grpc_json_parse_string(service_config_choice_json); + if (choices_json == NULL || choices_json->type != GRPC_JSON_ARRAY) { + gpr_log(GPR_ERROR, "cannot parse service config JSON string"); + return NULL; + } + char *service_config = NULL; + for (grpc_json *choice = choices_json->child; choice != NULL; + choice = choice->next) { + if (choice->type != GRPC_JSON_OBJECT) { + gpr_log(GPR_ERROR, "cannot parse service config JSON string"); + break; + } + grpc_json *service_config_json = NULL; + for (grpc_json *field = choice->child; field != NULL; field = field->next) { + // Check client language, if specified. + if (strcmp(field->key, "clientLanguage") == 0) { + if (field->type != GRPC_JSON_ARRAY || + !value_in_json_array(field, "c++")) { + service_config_json = NULL; + break; + } + } + // Check client hostname, if specified. + if (strcmp(field->key, "clientHostname") == 0) { + char *hostname = grpc_gethostname(); + if (hostname == NULL || field->type != GRPC_JSON_ARRAY || + !value_in_json_array(field, hostname)) { + service_config_json = NULL; + break; + } + } + // Check percentage, if specified. + if (strcmp(field->key, "percentage") == 0) { + if (field->type != GRPC_JSON_NUMBER) { + service_config_json = NULL; + break; + } + int random_pct = rand() % 100; + int percentage; + if (sscanf(field->value, "%d", &percentage) != 1 || + random_pct > percentage) { + service_config_json = NULL; + break; + } + } + // Save service config. + if (strcmp(field->key, "serviceConfig") == 0) { + if (field->type == GRPC_JSON_OBJECT) { + service_config_json = field; + } + } + } + if (service_config_json != NULL) { + service_config = grpc_json_dump_to_string(service_config_json, 0); + break; + } + } + grpc_json_destroy(choices_json); + return service_config; +} + static void dns_ares_on_resolved_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { ares_dns_resolver *r = arg; @@ -152,8 +233,40 @@ static void dns_ares_on_resolved_locked(grpc_exec_ctx *exec_ctx, void *arg, r->resolving = false; r->pending_request = NULL; if (r->lb_addresses != NULL) { - grpc_arg new_arg = grpc_lb_addresses_create_channel_arg(r->lb_addresses); - result = grpc_channel_args_copy_and_add(r->channel_args, &new_arg, 1); + static const char *args_to_remove[2]; + size_t num_args_to_remove = 0; + grpc_arg new_args[3]; + size_t num_args_to_add = 0; + new_args[num_args_to_add++] = + grpc_lb_addresses_create_channel_arg(r->lb_addresses); + grpc_service_config *service_config = NULL; + char *service_config_string = NULL; + if (r->service_config_json != NULL) { + service_config_string = choose_service_config(r->service_config_json); + gpr_free(r->service_config_json); + if (service_config_string != NULL) { + gpr_log(GPR_INFO, "selected service config choice: %s", + service_config_string); + args_to_remove[num_args_to_remove++] = GRPC_ARG_SERVICE_CONFIG; + new_args[num_args_to_add++] = grpc_channel_arg_string_create( + GRPC_ARG_SERVICE_CONFIG, service_config_string); + service_config = grpc_service_config_create(service_config_string); + if (service_config != NULL) { + const char *lb_policy_name = + grpc_service_config_get_lb_policy_name(service_config); + if (lb_policy_name != NULL) { + args_to_remove[num_args_to_remove++] = GRPC_ARG_LB_POLICY_NAME; + new_args[num_args_to_add++] = grpc_channel_arg_string_create( + GRPC_ARG_LB_POLICY_NAME, (char *)lb_policy_name); + } + } + } + } + result = grpc_channel_args_copy_and_add_and_remove( + r->channel_args, args_to_remove, num_args_to_remove, new_args, + num_args_to_add); + if (service_config != NULL) grpc_service_config_destroy(service_config); + gpr_free(service_config_string); grpc_lb_addresses_destroy(exec_ctx, r->lb_addresses); } else { const char *msg = grpc_error_string(error); @@ -207,10 +320,12 @@ static void dns_ares_start_resolving_locked(grpc_exec_ctx *exec_ctx, GPR_ASSERT(!r->resolving); r->resolving = true; r->lb_addresses = NULL; + r->service_config_json = NULL; r->pending_request = grpc_dns_lookup_ares( exec_ctx, r->dns_server, r->name_to_resolve, r->default_port, r->interested_parties, &r->dns_ares_on_resolved_locked, &r->lb_addresses, - true /* check_grpclb */); + true /* check_grpclb */, + r->request_service_config ? &r->service_config_json : NULL); } static void dns_ares_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx, @@ -256,6 +371,10 @@ static grpc_resolver *dns_ares_create(grpc_exec_ctx *exec_ctx, r->name_to_resolve = gpr_strdup(path); r->default_port = gpr_strdup(default_port); r->channel_args = grpc_channel_args_copy(args->args); + const grpc_arg *arg = grpc_channel_args_find( + r->channel_args, GRPC_ARG_SERVICE_CONFIG_DISABLE_RESOLUTION); + r->request_service_config = !grpc_channel_arg_get_integer( + arg, (grpc_integer_options){false, false, true}); r->interested_parties = grpc_pollset_set_create(); if (args->pollset_set != NULL) { grpc_pollset_set_add_pollset_set(exec_ctx, r->interested_parties, diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c index 6ec3790a5f..e65723a63b 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c @@ -54,6 +54,8 @@ struct grpc_ares_request { grpc_closure *on_done; /** the pointer to receive the resolved addresses */ grpc_lb_addresses **lb_addrs_out; + /** the pointer to receive the service config in JSON */ + char **service_config_json_out; /** the evernt driver used by this request */ grpc_ares_ev_driver *ev_driver; /** number of ongoing queries */ @@ -266,10 +268,68 @@ static void on_srv_query_done_cb(void *arg, int status, int timeouts, grpc_exec_ctx_finish(&exec_ctx); } +static const char g_service_config_attribute_prefix[] = "grpc_config="; + +static void on_txt_done_cb(void *arg, int status, int timeouts, + unsigned char *buf, int len) { + gpr_log(GPR_DEBUG, "on_txt_done_cb"); + char *error_msg; + grpc_ares_request *r = (grpc_ares_request *)arg; + gpr_mu_lock(&r->mu); + if (status != ARES_SUCCESS) goto fail; + struct ares_txt_ext *reply = NULL; + status = ares_parse_txt_reply_ext(buf, len, &reply); + if (status != ARES_SUCCESS) goto fail; + // Find service config in TXT record. + const size_t prefix_len = sizeof(g_service_config_attribute_prefix) - 1; + struct ares_txt_ext *result; + for (result = reply; result != NULL; result = result->next) { + if (result->record_start && + memcmp(result->txt, g_service_config_attribute_prefix, prefix_len) == + 0) { + break; + } + } + // Found a service config record. + if (result != NULL) { + size_t service_config_len = result->length - prefix_len; + *r->service_config_json_out = gpr_malloc(service_config_len + 1); + memcpy(*r->service_config_json_out, result->txt + prefix_len, + service_config_len); + for (result = result->next; result != NULL && !result->record_start; + result = result->next) { + *r->service_config_json_out = gpr_realloc( + *r->service_config_json_out, service_config_len + result->length + 1); + memcpy(*r->service_config_json_out + service_config_len, result->txt, + result->length); + service_config_len += result->length; + } + (*r->service_config_json_out)[service_config_len] = '\0'; + gpr_log(GPR_INFO, "found service config: %s", *r->service_config_json_out); + } + // Clean up. + ares_free_data(reply); + goto done; +fail: + gpr_asprintf(&error_msg, "C-ares TXT lookup status is not ARES_SUCCESS: %s", + ares_strerror(status)); + grpc_error *error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_msg); + gpr_free(error_msg); + if (r->error == GRPC_ERROR_NONE) { + r->error = error; + } else { + r->error = grpc_error_add_child(error, r->error); + } +done: + gpr_mu_unlock(&r->mu); + grpc_ares_request_unref(NULL, r); +} + static grpc_ares_request *grpc_dns_lookup_ares_impl( grpc_exec_ctx *exec_ctx, const char *dns_server, const char *name, const char *default_port, grpc_pollset_set *interested_parties, - grpc_closure *on_done, grpc_lb_addresses **addrs, bool check_grpclb) { + grpc_closure *on_done, grpc_lb_addresses **addrs, bool check_grpclb, + char **service_config_json) { grpc_error *error = GRPC_ERROR_NONE; /* TODO(zyc): Enable tracing after #9603 is checked in */ /* if (grpc_dns_trace) { @@ -300,11 +360,12 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( error = grpc_ares_ev_driver_create(&ev_driver, interested_parties); if (error != GRPC_ERROR_NONE) goto error_cleanup; - grpc_ares_request *r = gpr_malloc(sizeof(grpc_ares_request)); + grpc_ares_request *r = gpr_zalloc(sizeof(grpc_ares_request)); gpr_mu_init(&r->mu); r->ev_driver = ev_driver; r->on_done = on_done; r->lb_addrs_out = addrs; + r->service_config_json_out = service_config_json; r->success = false; r->error = GRPC_ERROR_NONE; ares_channel *channel = grpc_ares_ev_driver_get_channel(r->ev_driver); @@ -315,13 +376,17 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( grpc_resolved_address addr; if (grpc_parse_ipv4_hostport(dns_server, &addr, false /* log_errors */)) { r->dns_server_addr.family = AF_INET; - memcpy(&r->dns_server_addr.addr.addr4, addr.addr, addr.len); + struct sockaddr_in *in = (struct sockaddr_in *)addr.addr; + memcpy(&r->dns_server_addr.addr.addr4, &in->sin_addr, + sizeof(struct in_addr)); r->dns_server_addr.tcp_port = grpc_sockaddr_get_port(&addr); r->dns_server_addr.udp_port = grpc_sockaddr_get_port(&addr); } else if (grpc_parse_ipv6_hostport(dns_server, &addr, false /* log_errors */)) { r->dns_server_addr.family = AF_INET6; - memcpy(&r->dns_server_addr.addr.addr6, addr.addr, addr.len); + struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)addr.addr; + memcpy(&r->dns_server_addr.addr.addr6, &in6->sin6_addr, + sizeof(struct in6_addr)); r->dns_server_addr.tcp_port = grpc_sockaddr_get_port(&addr); r->dns_server_addr.udp_port = grpc_sockaddr_get_port(&addr); } else { @@ -342,8 +407,6 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( goto error_cleanup; } } - // An extra reference is put here to avoid destroying the request in - // on_done_cb before calling grpc_ares_ev_driver_start. gpr_ref_init(&r->pending_queries, 1); if (grpc_ipv6_loopback_available()) { grpc_ares_hostbyname_request *hr = create_hostbyname_request( @@ -362,6 +425,10 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( r); gpr_free(service_name); } + if (service_config_json != NULL) { + grpc_ares_request_ref(r); + ares_search(*channel, hr->host, ns_c_in, ns_t_txt, on_txt_done_cb, r); + } /* TODO(zyc): Handle CNAME records here. */ grpc_ares_ev_driver_start(exec_ctx, r->ev_driver); grpc_ares_request_unref(exec_ctx, r); @@ -379,8 +446,8 @@ error_cleanup: grpc_ares_request *(*grpc_dns_lookup_ares)( grpc_exec_ctx *exec_ctx, const char *dns_server, const char *name, const char *default_port, grpc_pollset_set *interested_parties, - grpc_closure *on_done, grpc_lb_addresses **addrs, - bool check_grpclb) = grpc_dns_lookup_ares_impl; + grpc_closure *on_done, grpc_lb_addresses **addrs, bool check_grpclb, + char **service_config_json) = grpc_dns_lookup_ares_impl; void grpc_cancel_ares_request(grpc_exec_ctx *exec_ctx, grpc_ares_request *r) { if (grpc_dns_lookup_ares == grpc_dns_lookup_ares_impl) { @@ -465,7 +532,8 @@ static void grpc_resolve_address_ares_impl(grpc_exec_ctx *exec_ctx, grpc_schedule_on_exec_ctx); grpc_dns_lookup_ares(exec_ctx, NULL /* dns_server */, name, default_port, interested_parties, &r->on_dns_lookup_done, &r->lb_addrs, - false /* check_grpclb */); + false /* check_grpclb */, + NULL /* service_config_json */); } void (*grpc_resolve_address_ares)( diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h index 5d2d6c993b..108333047d 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h @@ -27,29 +27,30 @@ typedef struct grpc_ares_request grpc_ares_request; -/* Asynchronously resolve addr. Use \a default_port if a port isn't designated - in addr, otherwise use the port in addr. grpc_ares_init() must be called at - least once before this function. \a on_done may be called directly in this - function without being scheduled with \a exec_ctx, it must not try to acquire - locks that are being held by the caller. */ +/* Asynchronously resolve \a name. Use \a default_port if a port isn't + designated in \a name, otherwise use the port in \a name. grpc_ares_init() + must be called at least once before this function. \a on_done may be + called directly in this function without being scheduled with \a exec_ctx, + so it must not try to acquire locks that are being held by the caller. */ extern void (*grpc_resolve_address_ares)(grpc_exec_ctx *exec_ctx, - const char *addr, + const char *name, const char *default_port, grpc_pollset_set *interested_parties, grpc_closure *on_done, grpc_resolved_addresses **addresses); -/* Asynchronously resolve addr. It will try to resolve grpclb SRV records in +/* Asynchronously resolve \a name. It will try to resolve grpclb SRV records in addition to the normal address records. For normal address records, it uses - \a default_port if a port isn't designated in \a addr, otherwise it uses the - port in \a addr. grpc_ares_init() must be called at least once before this + \a default_port if a port isn't designated in \a name, otherwise it uses the + port in \a name. grpc_ares_init() must be called at least once before this function. \a on_done may be called directly in this function without being - scheduled with \a exec_ctx, it must not try to acquire locks that are being - held by the caller. */ + scheduled with \a exec_ctx, so it must not try to acquire locks that are + being held by the caller. */ extern grpc_ares_request *(*grpc_dns_lookup_ares)( - grpc_exec_ctx *exec_ctx, const char *dns_server, const char *addr, + grpc_exec_ctx *exec_ctx, const char *dns_server, const char *name, const char *default_port, grpc_pollset_set *interested_parties, - grpc_closure *on_done, grpc_lb_addresses **addresses, bool check_grpclb); + grpc_closure *on_done, grpc_lb_addresses **addresses, bool check_grpclb, + char **service_config_json); /* Cancel the pending grpc_ares_request \a request */ void grpc_cancel_ares_request(grpc_exec_ctx *exec_ctx, diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.c index b67636a3e4..f2587c4520 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.c @@ -28,15 +28,16 @@ struct grpc_ares_request { static grpc_ares_request *grpc_dns_lookup_ares_impl( grpc_exec_ctx *exec_ctx, const char *dns_server, const char *name, const char *default_port, grpc_pollset_set *interested_parties, - grpc_closure *on_done, grpc_lb_addresses **addrs, bool check_grpclb) { + grpc_closure *on_done, grpc_lb_addresses **addrs, bool check_grpclb, + char **service_config_json) { return NULL; } grpc_ares_request *(*grpc_dns_lookup_ares)( grpc_exec_ctx *exec_ctx, const char *dns_server, const char *name, const char *default_port, grpc_pollset_set *interested_parties, - grpc_closure *on_done, grpc_lb_addresses **addrs, - bool check_grpclb) = grpc_dns_lookup_ares_impl; + grpc_closure *on_done, grpc_lb_addresses **addrs, bool check_grpclb, + char **service_config_json) = grpc_dns_lookup_ares_impl; void grpc_cancel_ares_request(grpc_exec_ctx *exec_ctx, grpc_ares_request *r) {} diff --git a/src/core/lib/iomgr/ev_epoll1_linux.c b/src/core/lib/iomgr/ev_epoll1_linux.c index dc48d73df9..90e0ce36cd 100644 --- a/src/core/lib/iomgr/ev_epoll1_linux.c +++ b/src/core/lib/iomgr/ev_epoll1_linux.c @@ -237,28 +237,41 @@ static grpc_fd *fd_create(int fd, const char *name) { static int fd_wrapped_fd(grpc_fd *fd) { return fd->fd; } -/* Might be called multiple times */ -static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) { +/* if 'releasing_fd' is true, it means that we are going to detach the internal + * fd from grpc_fd structure (i.e which means we should not be calling + * shutdown() syscall on that fd) */ +static void fd_shutdown_internal(grpc_exec_ctx *exec_ctx, grpc_fd *fd, + grpc_error *why, bool releasing_fd) { if (grpc_lfev_set_shutdown(exec_ctx, &fd->read_closure, GRPC_ERROR_REF(why))) { - shutdown(fd->fd, SHUT_RDWR); + if (!releasing_fd) { + shutdown(fd->fd, SHUT_RDWR); + } grpc_lfev_set_shutdown(exec_ctx, &fd->write_closure, GRPC_ERROR_REF(why)); } GRPC_ERROR_UNREF(why); } +/* Might be called multiple times */ +static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) { + fd_shutdown_internal(exec_ctx, fd, why, false); +} + static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure *on_done, int *release_fd, bool already_closed, const char *reason) { grpc_error *error = GRPC_ERROR_NONE; + bool is_release_fd = (release_fd != NULL); if (!grpc_lfev_is_shutdown(&fd->read_closure)) { - fd_shutdown(exec_ctx, fd, GRPC_ERROR_CREATE_FROM_COPIED_STRING(reason)); + fd_shutdown_internal(exec_ctx, fd, + GRPC_ERROR_CREATE_FROM_COPIED_STRING(reason), + is_release_fd); } /* If release_fd is not NULL, we should be relinquishing control of the file descriptor fd->fd (but we still own the grpc_fd structure). */ - if (release_fd != NULL) { + if (is_release_fd) { *release_fd = fd->fd; } else if (!already_closed) { close(fd->fd); diff --git a/src/core/lib/iomgr/gethostname.h b/src/core/lib/iomgr/gethostname.h new file mode 100644 index 0000000000..9c6b9d8d42 --- /dev/null +++ b/src/core/lib/iomgr/gethostname.h @@ -0,0 +1,26 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#ifndef GRPC_CORE_LIB_IOMGR_GETHOSTNAME_H +#define GRPC_CORE_LIB_IOMGR_GETHOSTNAME_H + +// Returns the hostname of the local machine. +// Caller takes ownership of result. +char *grpc_gethostname(); + +#endif /* GRPC_CORE_LIB_IOMGR_GETHOSTNAME_H */ diff --git a/src/core/lib/iomgr/gethostname_fallback.c b/src/core/lib/iomgr/gethostname_fallback.c new file mode 100644 index 0000000000..6229461568 --- /dev/null +++ b/src/core/lib/iomgr/gethostname_fallback.c @@ -0,0 +1,27 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#include "src/core/lib/iomgr/port.h" + +#ifdef GRPC_GETHOSTNAME_FALLBACK + +#include <stddef.h> + +char *grpc_gethostname() { return NULL; } + +#endif // GRPC_GETHOSTNAME_FALLBACK diff --git a/src/core/lib/iomgr/gethostname_host_name_max.c b/src/core/lib/iomgr/gethostname_host_name_max.c new file mode 100644 index 0000000000..4d0511412e --- /dev/null +++ b/src/core/lib/iomgr/gethostname_host_name_max.c @@ -0,0 +1,37 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#include "src/core/lib/iomgr/port.h" + +#ifdef GRPC_POSIX_HOST_NAME_MAX + +#include <limits.h> +#include <unistd.h> + +#include <grpc/support/alloc.h> + +char *grpc_gethostname() { + char *hostname = (char *)gpr_malloc(HOST_NAME_MAX); + if (gethostname(hostname, HOST_NAME_MAX) != 0) { + gpr_free(hostname); + return NULL; + } + return hostname; +} + +#endif // GRPC_POSIX_HOST_NAME_MAX diff --git a/src/core/lib/iomgr/gethostname_sysconf.c b/src/core/lib/iomgr/gethostname_sysconf.c new file mode 100644 index 0000000000..51bac5d69d --- /dev/null +++ b/src/core/lib/iomgr/gethostname_sysconf.c @@ -0,0 +1,37 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#include "src/core/lib/iomgr/port.h" + +#ifdef GRPC_POSIX_SYSCONF + +#include <unistd.h> + +#include <grpc/support/alloc.h> + +char *grpc_gethostname() { + size_t host_name_max = (size_t)sysconf(_SC_HOST_NAME_MAX); + char *hostname = (char *)gpr_malloc(host_name_max); + if (gethostname(hostname, host_name_max) != 0) { + gpr_free(hostname); + return NULL; + } + return hostname; +} + +#endif // GRPC_POSIX_SYSCONF diff --git a/src/core/lib/iomgr/port.h b/src/core/lib/iomgr/port.h index c12058f890..42033d0ba4 100644 --- a/src/core/lib/iomgr/port.h +++ b/src/core/lib/iomgr/port.h @@ -59,6 +59,7 @@ #define GRPC_HAVE_MSG_NOSIGNAL 1 #define GRPC_HAVE_UNIX_SOCKET 1 #define GRPC_LINUX_MULTIPOLL_WITH_EPOLL 1 +#define GRPC_POSIX_HOST_NAME_MAX 1 #define GRPC_POSIX_SOCKET 1 #define GRPC_POSIX_SOCKETADDR 1 #define GRPC_POSIX_WAKEUP_FD 1 @@ -93,6 +94,7 @@ #define GRPC_POSIX_SOCKET 1 #define GRPC_POSIX_SOCKETADDR 1 #define GRPC_POSIX_SOCKETUTILS 1 +#define GRPC_POSIX_SYSCONF 1 #define GRPC_POSIX_WAKEUP_FD 1 #define GRPC_TIMER_USE_GENERIC 1 #elif defined(GPR_FREEBSD) @@ -125,4 +127,11 @@ #error Must define exactly one of GRPC_POSIX_SOCKET, GRPC_WINSOCK_SOCKET, GPR_CUSTOM_SOCKET #endif +#if defined(GRPC_POSIX_HOST_NAME_MAX) && defined(GRPC_POSIX_SYSCONF) +#error "Cannot define both GRPC_POSIX_HOST_NAME_MAX and GRPC_POSIX_SYSCONF" +#endif +#if !defined(GRPC_POSIX_HOST_NAME_MAX) && !defined(GRPC_POSIX_SYSCONF) +#define GRPC_GETHOSTNAME_FALLBACK 1 +#endif + #endif /* GRPC_CORE_LIB_IOMGR_PORT_H */ diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_common.c b/src/core/lib/iomgr/tcp_server_utils_posix_common.c index dbb43186bd..ad535bc43e 100644 --- a/src/core/lib/iomgr/tcp_server_utils_posix_common.c +++ b/src/core/lib/iomgr/tcp_server_utils_posix_common.c @@ -39,7 +39,7 @@ #define MIN_SAFE_ACCEPT_QUEUE_SIZE 100 -static gpr_once s_init_max_accept_queue_size; +static gpr_once s_init_max_accept_queue_size = GPR_ONCE_INIT; static int s_max_accept_queue_size; /* get max listen queue size on linux */ diff --git a/src/core/lib/surface/alarm.c b/src/core/lib/surface/alarm.c index 55934964f3..7d60b1de17 100644 --- a/src/core/lib/surface/alarm.c +++ b/src/core/lib/surface/alarm.c @@ -15,6 +15,7 @@ * limitations under the License. * */ +#include "src/core/lib/surface/alarm_internal.h" #include <grpc/grpc.h> #include <grpc/support/alloc.h> @@ -22,7 +23,13 @@ #include "src/core/lib/iomgr/timer.h" #include "src/core/lib/surface/completion_queue.h" +#ifndef NDEBUG +grpc_tracer_flag grpc_trace_alarm_refcount = + GRPC_TRACER_INITIALIZER(false, "alarm_refcount"); +#endif + struct grpc_alarm { + gpr_refcount refs; grpc_timer alarm; grpc_closure on_alarm; grpc_cq_completion completion; @@ -32,13 +39,58 @@ struct grpc_alarm { void *tag; }; -static void do_nothing_end_completion(grpc_exec_ctx *exec_ctx, void *arg, - grpc_cq_completion *c) {} +static void alarm_ref(grpc_alarm *alarm) { gpr_ref(&alarm->refs); } + +static void alarm_unref(grpc_alarm *alarm) { + if (gpr_unref(&alarm->refs)) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + GRPC_CQ_INTERNAL_UNREF(&exec_ctx, alarm->cq, "alarm"); + grpc_exec_ctx_finish(&exec_ctx); + gpr_free(alarm); + } +} + +#ifndef NDEBUG +static void alarm_ref_dbg(grpc_alarm *alarm, const char *reason, + const char *file, int line) { + if (GRPC_TRACER_ON(grpc_trace_alarm_refcount)) { + gpr_atm val = gpr_atm_no_barrier_load(&alarm->refs.count); + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, + "Alarm:%p ref %" PRIdPTR " -> %" PRIdPTR " %s", alarm, val, + val + 1, reason); + } + + alarm_ref(alarm); +} + +static void alarm_unref_dbg(grpc_alarm *alarm, const char *reason, + const char *file, int line) { + if (GRPC_TRACER_ON(grpc_trace_alarm_refcount)) { + gpr_atm val = gpr_atm_no_barrier_load(&alarm->refs.count); + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, + "Alarm:%p Unref %" PRIdPTR " -> %" PRIdPTR " %s", alarm, val, + val - 1, reason); + } + + alarm_unref(alarm); +} +#endif + +static void alarm_end_completion(grpc_exec_ctx *exec_ctx, void *arg, + grpc_cq_completion *c) { + grpc_alarm *alarm = arg; + GRPC_ALARM_UNREF(alarm, "dequeue-end-op"); +} static void alarm_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_alarm *alarm = arg; - grpc_cq_end_op(exec_ctx, alarm->cq, alarm->tag, error, - do_nothing_end_completion, NULL, &alarm->completion); + + /* We are queuing an op on completion queue. This means, the alarm's structure + cannot be destroyed until the op is dequeued. Adding an extra ref + here and unref'ing when the op is dequeued will achieve this */ + GRPC_ALARM_REF(alarm, "queue-end-op"); + grpc_cq_end_op(exec_ctx, alarm->cq, alarm->tag, error, alarm_end_completion, + (void *)alarm, &alarm->completion); } grpc_alarm *grpc_alarm_create(grpc_completion_queue *cq, gpr_timespec deadline, @@ -46,6 +98,14 @@ grpc_alarm *grpc_alarm_create(grpc_completion_queue *cq, gpr_timespec deadline, grpc_alarm *alarm = gpr_malloc(sizeof(grpc_alarm)); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + gpr_ref_init(&alarm->refs, 1); + +#ifndef NDEBUG + if (GRPC_TRACER_ON(grpc_trace_alarm_refcount)) { + gpr_log(GPR_DEBUG, "Alarm:%p created (ref: 1)", alarm); + } +#endif + GRPC_CQ_INTERNAL_REF(cq, "alarm"); alarm->cq = cq; alarm->tag = tag; @@ -67,9 +127,6 @@ void grpc_alarm_cancel(grpc_alarm *alarm) { } void grpc_alarm_destroy(grpc_alarm *alarm) { - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_alarm_cancel(alarm); - GRPC_CQ_INTERNAL_UNREF(&exec_ctx, alarm->cq, "alarm"); - gpr_free(alarm); - grpc_exec_ctx_finish(&exec_ctx); + GRPC_ALARM_UNREF(alarm, "alarm_destroy"); } diff --git a/src/core/lib/surface/alarm_internal.h b/src/core/lib/surface/alarm_internal.h new file mode 100644 index 0000000000..7f2126c5c9 --- /dev/null +++ b/src/core/lib/surface/alarm_internal.h @@ -0,0 +1,40 @@ +/* + * + * Copyright 2015-2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#ifndef GRPC_CORE_LIB_SURFACE_ALARM_INTERNAL_H +#define GRPC_CORE_LIB_SURFACE_ALARM_INTERNAL_H + +#include <grpc/support/log.h> +#include "src/core/lib/debug/trace.h" + +#ifndef NDEBUG + +extern grpc_tracer_flag grpc_trace_alarm_refcount; + +#define GRPC_ALARM_REF(a, reason) alarm_ref_dbg(a, reason, __FILE__, __LINE__) +#define GRPC_ALARM_UNREF(a, reason) \ + alarm_unref_dbg(a, reason, __FILE__, __LINE__) + +#else /* !defined(NDEBUG) */ + +#define GRPC_ALARM_REF(a, reason) alarm_ref(a) +#define GRPC_ALARM_UNREF(a, reason) alarm_unref(a) + +#endif /* defined(NDEBUG) */ + +#endif /* GRPC_CORE_LIB_SURFACE_ALARM_INTERNAL_H */ diff --git a/src/core/lib/surface/init.c b/src/core/lib/surface/init.c index db111e597f..d199ac060e 100644 --- a/src/core/lib/surface/init.c +++ b/src/core/lib/surface/init.c @@ -36,6 +36,7 @@ #include "src/core/lib/iomgr/resource_quota.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/slice_internal.h" +#include "src/core/lib/surface/alarm_internal.h" #include "src/core/lib/surface/api_trace.h" #include "src/core/lib/surface/call.h" #include "src/core/lib/surface/channel_init.h" @@ -135,6 +136,7 @@ void grpc_init(void) { grpc_register_tracer(&grpc_call_error_trace); #ifndef NDEBUG grpc_register_tracer(&grpc_trace_pending_tags); + grpc_register_tracer(&grpc_trace_alarm_refcount); grpc_register_tracer(&grpc_trace_cq_refcount); grpc_register_tracer(&grpc_trace_closure); grpc_register_tracer(&grpc_trace_error_refcount); diff --git a/src/cpp/util/slice_cc.cc b/src/cpp/util/slice_cc.cc index 56e0328b94..486d0cdf0e 100644 --- a/src/cpp/util/slice_cc.cc +++ b/src/cpp/util/slice_cc.cc @@ -17,6 +17,7 @@ */ #include <grpc++/support/slice.h> +#include <grpc/slice.h> namespace grpc { @@ -43,4 +44,10 @@ Slice::Slice(const void* buf, size_t len, StaticSlice) Slice::Slice(const Slice& other) : slice_(grpc_slice_ref(other.slice_)) {} +Slice::Slice(void* buf, size_t len, void (*destroy)(void*), void* user_data) + : slice_(grpc_slice_new_with_user_data(buf, len, destroy, user_data)) {} + +Slice::Slice(void* buf, size_t len, void (*destroy)(void*, size_t)) + : slice_(grpc_slice_new_with_len(buf, len, destroy)) {} + } // namespace grpc diff --git a/src/php/ext/grpc/call.c b/src/php/ext/grpc/call.c index 2f67e5cee7..c4997f720d 100644 --- a/src/php/ext/grpc/call.c +++ b/src/php/ext/grpc/call.c @@ -214,10 +214,12 @@ PHP_METHOD(Call, __construct) { return; } wrapped_grpc_channel *channel = Z_WRAPPED_GRPC_CHANNEL_P(channel_obj); - if (channel->wrapped == NULL) { + gpr_mu_lock(&channel->wrapper->mu); + if (channel->wrapper->wrapped == NULL) { zend_throw_exception(spl_ce_InvalidArgumentException, "Call cannot be constructed from a closed Channel", 1 TSRMLS_CC); + gpr_mu_unlock(&channel->wrapper->mu); return; } add_property_zval(getThis(), "channel", channel_obj); @@ -226,13 +228,15 @@ PHP_METHOD(Call, __construct) { grpc_slice host_slice = host_override != NULL ? grpc_slice_from_copied_string(host_override) : grpc_empty_slice(); call->wrapped = - grpc_channel_create_call(channel->wrapped, NULL, GRPC_PROPAGATE_DEFAULTS, + grpc_channel_create_call(channel->wrapper->wrapped, NULL, + GRPC_PROPAGATE_DEFAULTS, completion_queue, method_slice, host_override != NULL ? &host_slice : NULL, deadline->wrapped, NULL); grpc_slice_unref(method_slice); grpc_slice_unref(host_slice); call->owned = true; + gpr_mu_unlock(&channel->wrapper->mu); } /** diff --git a/src/php/ext/grpc/call_credentials.c b/src/php/ext/grpc/call_credentials.c index a990206c08..f46091d709 100644 --- a/src/php/ext/grpc/call_credentials.c +++ b/src/php/ext/grpc/call_credentials.c @@ -109,8 +109,8 @@ PHP_METHOD(CallCredentials, createFromPlugin) { zend_fcall_info *fci; zend_fcall_info_cache *fci_cache; - fci = (zend_fcall_info *)emalloc(sizeof(zend_fcall_info)); - fci_cache = (zend_fcall_info_cache *)emalloc(sizeof(zend_fcall_info_cache)); + fci = (zend_fcall_info *)malloc(sizeof(zend_fcall_info)); + fci_cache = (zend_fcall_info_cache *)malloc(sizeof(zend_fcall_info_cache)); memset(fci, 0, sizeof(zend_fcall_info)); memset(fci_cache, 0, sizeof(zend_fcall_info_cache)); @@ -123,7 +123,7 @@ PHP_METHOD(CallCredentials, createFromPlugin) { } plugin_state *state; - state = (plugin_state *)emalloc(sizeof(plugin_state)); + state = (plugin_state *)malloc(sizeof(plugin_state)); memset(state, 0, sizeof(plugin_state)); /* save the user provided PHP callback function */ @@ -210,13 +210,13 @@ void plugin_get_metadata(void *ptr, grpc_auth_metadata_context context, /* Cleanup function for plugin creds API */ void plugin_destroy_state(void *ptr) { plugin_state *state = (plugin_state *)ptr; - efree(state->fci); - efree(state->fci_cache); + free(state->fci); + free(state->fci_cache); #if PHP_MAJOR_VERSION < 7 PHP_GRPC_FREE_STD_ZVAL(state->fci->params); PHP_GRPC_FREE_STD_ZVAL(state->fci->retval); #endif - efree(state); + free(state); } ZEND_BEGIN_ARG_INFO_EX(arginfo_createComposite, 0, 0, 2) diff --git a/src/php/ext/grpc/channel.c b/src/php/ext/grpc/channel.c index 6c432d2818..f1187e8722 100644 --- a/src/php/ext/grpc/channel.c +++ b/src/php/ext/grpc/channel.c @@ -25,6 +25,13 @@ #include <php.h> #include <php_ini.h> #include <ext/standard/info.h> +#include <ext/standard/php_var.h> +#include <ext/standard/sha1.h> +#if PHP_MAJOR_VERSION < 7 +#include <ext/standard/php_smart_str.h> +#else +#include <zend_smart_str.h> +#endif #include <ext/spl/spl_exceptions.h> #include "php_grpc.h" @@ -44,11 +51,25 @@ zend_class_entry *grpc_ce_channel; #if PHP_MAJOR_VERSION >= 7 static zend_object_handlers channel_ce_handlers; #endif +static gpr_mu global_persistent_list_mu; +int le_plink; /* Frees and destroys an instance of wrapped_grpc_channel */ PHP_GRPC_FREE_WRAPPED_FUNC_START(wrapped_grpc_channel) - if (p->wrapped != NULL) { - grpc_channel_destroy(p->wrapped); + if (p->wrapper != NULL) { + gpr_mu_lock(&p->wrapper->mu); + if (p->wrapper->wrapped != NULL) { + php_grpc_zend_resource *rsrc; + php_grpc_int key_len = strlen(p->wrapper->key); + // only destroy the channel here if not found in the persistent list + gpr_mu_lock(&global_persistent_list_mu); + if (!(PHP_GRPC_PERSISTENT_LIST_FIND(&EG(persistent_list), p->wrapper->key, + key_len, rsrc))) { + grpc_channel_destroy(p->wrapper->wrapped); + } + gpr_mu_unlock(&global_persistent_list_mu); + } + gpr_mu_unlock(&p->wrapper->mu); } PHP_GRPC_FREE_WRAPPED_FUNC_END() @@ -62,15 +83,15 @@ php_grpc_zend_object create_wrapped_grpc_channel(zend_class_entry *class_type PHP_GRPC_FREE_CLASS_OBJECT(wrapped_grpc_channel, channel_ce_handlers); } -void php_grpc_read_args_array(zval *args_array, - grpc_channel_args *args TSRMLS_DC) { +int php_grpc_read_args_array(zval *args_array, + grpc_channel_args *args TSRMLS_DC) { HashTable *array_hash; int args_index; array_hash = Z_ARRVAL_P(args_array); if (!array_hash) { zend_throw_exception(spl_ce_InvalidArgumentException, "array_hash is NULL", 1 TSRMLS_CC); - return; + return FAILURE; } args->num_args = zend_hash_num_elements(array_hash); args->args = ecalloc(args->num_args, sizeof(grpc_arg)); @@ -84,7 +105,7 @@ void php_grpc_read_args_array(zval *args_array, if (key_type != HASH_KEY_IS_STRING) { zend_throw_exception(spl_ce_InvalidArgumentException, "args keys must be strings", 1 TSRMLS_CC); - return; + return FAILURE; } args->args[args_index].key = key; switch (Z_TYPE_P(data)) { @@ -99,16 +120,78 @@ void php_grpc_read_args_array(zval *args_array, default: zend_throw_exception(spl_ce_InvalidArgumentException, "args values must be int or string", 1 TSRMLS_CC); - return; + return FAILURE; } args_index++; PHP_GRPC_HASH_FOREACH_END() + return SUCCESS; +} + +void generate_sha1_str(char *sha1str, char *str, php_grpc_int len) { + PHP_SHA1_CTX context; + unsigned char digest[20]; + sha1str[0] = '\0'; + PHP_SHA1Init(&context); + PHP_GRPC_SHA1Update(&context, str, len); + PHP_SHA1Final(digest, &context); + make_sha1_digest(sha1str, digest); +} + +void create_channel( + wrapped_grpc_channel *channel, + char *target, + grpc_channel_args args, + wrapped_grpc_channel_credentials *creds) { + if (creds == NULL) { + channel->wrapper->wrapped = grpc_insecure_channel_create(target, &args, + NULL); + } else { + channel->wrapper->wrapped = + grpc_secure_channel_create(creds->wrapped, target, &args, NULL); + } + efree(args.args); +} + +void create_and_add_channel_to_persistent_list( + wrapped_grpc_channel *channel, + char *target, + grpc_channel_args args, + wrapped_grpc_channel_credentials *creds, + char *key, + php_grpc_int key_len) { + php_grpc_zend_resource new_rsrc; + channel_persistent_le_t *le; + // this links each persistent list entry to a destructor + new_rsrc.type = le_plink; + le = malloc(sizeof(channel_persistent_le_t)); + + create_channel(channel, target, args, creds); + + le->channel = channel->wrapper; + new_rsrc.ptr = le; + gpr_mu_lock(&global_persistent_list_mu); + PHP_GRPC_PERSISTENT_LIST_UPDATE(&EG(persistent_list), key, key_len, + (void *)&new_rsrc); + gpr_mu_unlock(&global_persistent_list_mu); } /** - * Construct an instance of the Channel class. If the $args array contains a - * "credentials" key mapping to a ChannelCredentials object, a secure channel - * will be created with those credentials. + * Construct an instance of the Channel class. + * + * By default, the underlying grpc_channel is "persistent". That is, given + * the same set of parameters passed to the constructor, the same underlying + * grpc_channel will be returned. + * + * If the $args array contains a "credentials" key mapping to a + * ChannelCredentials object, a secure channel will be created with those + * credentials. + * + * If the $args array contains a "force_new" key mapping to a boolean value + * of "true", a new underlying grpc_channel will be created regardless. If + * there are any opened channels on the same hostname, user must manually + * call close() on those dangling channels before the end of the PHP + * script. + * * @param string $target The hostname to associate with this channel * @param array $args_array The arguments to pass to the Channel */ @@ -121,6 +204,9 @@ PHP_METHOD(Channel, __construct) { grpc_channel_args args; HashTable *array_hash; wrapped_grpc_channel_credentials *creds = NULL; + php_grpc_zend_resource *rsrc; + bool force_new = false; + zval *force_new_obj = NULL; /* "sa" == 1 string, 1 array */ if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sa", &target, @@ -131,7 +217,7 @@ PHP_METHOD(Channel, __construct) { } array_hash = Z_ARRVAL_P(args_array); if (php_grpc_zend_hash_find(array_hash, "credentials", sizeof("credentials"), - (void **)&creds_obj) == SUCCESS) { + (void **)&creds_obj) == SUCCESS) { if (Z_TYPE_P(creds_obj) == IS_NULL) { creds = NULL; php_grpc_zend_hash_del(array_hash, "credentials", sizeof("credentials")); @@ -146,14 +232,82 @@ PHP_METHOD(Channel, __construct) { php_grpc_zend_hash_del(array_hash, "credentials", sizeof("credentials")); } } - php_grpc_read_args_array(args_array, &args TSRMLS_CC); - if (creds == NULL) { - channel->wrapped = grpc_insecure_channel_create(target, &args, NULL); + if (php_grpc_zend_hash_find(array_hash, "force_new", sizeof("force_new"), + (void **)&force_new_obj) == SUCCESS) { + if (PHP_GRPC_BVAL_IS_TRUE(force_new_obj)) { + force_new = true; + } + php_grpc_zend_hash_del(array_hash, "force_new", sizeof("force_new")); + } + + // parse the rest of the channel args array + if (php_grpc_read_args_array(args_array, &args TSRMLS_CC) == FAILURE) { + return; + } + + // Construct a hashkey for the persistent channel + // Currently, the hashkey contains 3 parts: + // 1. hostname + // 2. hash value of the channel args array (excluding "credentials" + // and "force_new") + // 3. (optional) hash value of the ChannelCredentials object + php_serialize_data_t var_hash; + smart_str buf = {0}; + PHP_VAR_SERIALIZE_INIT(var_hash); + PHP_GRPC_VAR_SERIALIZE(&buf, args_array, &var_hash); + PHP_VAR_SERIALIZE_DESTROY(var_hash); + + char sha1str[41]; + generate_sha1_str(sha1str, PHP_GRPC_SERIALIZED_BUF_STR(buf), + PHP_GRPC_SERIALIZED_BUF_LEN(buf)); + + php_grpc_int key_len = target_length + strlen(sha1str); + if (creds != NULL && creds->hashstr != NULL) { + key_len += strlen(creds->hashstr); + } + char *key = malloc(key_len + 1); + strcpy(key, target); + strcat(key, sha1str); + if (creds != NULL && creds->hashstr != NULL) { + strcat(key, creds->hashstr); + } + channel->wrapper = malloc(sizeof(grpc_channel_wrapper)); + channel->wrapper->key = key; + channel->wrapper->target = target; + channel->wrapper->args_hashstr = sha1str; + if (creds != NULL && creds->hashstr != NULL) { + channel->wrapper->creds_hashstr = creds->hashstr; + } + gpr_mu_init(&channel->wrapper->mu); + smart_str_free(&buf); + + if (force_new) { + php_grpc_delete_persistent_list_entry(key, key_len TSRMLS_CC); + } + + if (creds != NULL && creds->has_call_creds) { + // If the ChannelCredentials object was composed with a CallCredentials + // object, there is no way we can tell them apart. Do NOT persist + // them. They should be individually destroyed. + create_channel(channel, target, args, creds); + } else if (!(PHP_GRPC_PERSISTENT_LIST_FIND(&EG(persistent_list), key, + key_len, rsrc))) { + create_and_add_channel_to_persistent_list( + channel, target, args, creds, key, key_len); } else { - channel->wrapped = - grpc_secure_channel_create(creds->wrapped, target, &args, NULL); + // Found a previously stored channel in the persistent list + channel_persistent_le_t *le = (channel_persistent_le_t *)rsrc->ptr; + if (strcmp(target, le->channel->target) != 0 || + strcmp(sha1str, le->channel->args_hashstr) != 0 || + (creds != NULL && creds->hashstr != NULL && + strcmp(creds->hashstr, le->channel->creds_hashstr) != 0)) { + // somehow hash collision + create_and_add_channel_to_persistent_list( + channel, target, args, creds, key, key_len); + } else { + channel->wrapper = le->channel; + } } - efree(args.args); } /** @@ -162,7 +316,16 @@ PHP_METHOD(Channel, __construct) { */ PHP_METHOD(Channel, getTarget) { wrapped_grpc_channel *channel = Z_WRAPPED_GRPC_CHANNEL_P(getThis()); - PHP_GRPC_RETURN_STRING(grpc_channel_get_target(channel->wrapped), 1); + gpr_mu_lock(&channel->wrapper->mu); + if (channel->wrapper->wrapped == NULL) { + zend_throw_exception(spl_ce_RuntimeException, + "Channel already closed", 1 TSRMLS_CC); + gpr_mu_unlock(&channel->wrapper->mu); + return; + } + char *target = grpc_channel_get_target(channel->wrapper->wrapped); + gpr_mu_unlock(&channel->wrapper->mu); + PHP_GRPC_RETURN_STRING(target, 1); } /** @@ -172,6 +335,14 @@ PHP_METHOD(Channel, getTarget) { */ PHP_METHOD(Channel, getConnectivityState) { wrapped_grpc_channel *channel = Z_WRAPPED_GRPC_CHANNEL_P(getThis()); + gpr_mu_lock(&channel->wrapper->mu); + if (channel->wrapper->wrapped == NULL) { + zend_throw_exception(spl_ce_RuntimeException, + "Channel already closed", 1 TSRMLS_CC); + gpr_mu_unlock(&channel->wrapper->mu); + return; + } + bool try_to_connect = false; /* "|b" == 1 optional bool */ @@ -179,10 +350,18 @@ PHP_METHOD(Channel, getConnectivityState) { == FAILURE) { zend_throw_exception(spl_ce_InvalidArgumentException, "getConnectivityState expects a bool", 1 TSRMLS_CC); + gpr_mu_unlock(&channel->wrapper->mu); return; } - RETURN_LONG(grpc_channel_check_connectivity_state(channel->wrapped, - (int)try_to_connect)); + int state = grpc_channel_check_connectivity_state(channel->wrapper->wrapped, + (int)try_to_connect); + // this can happen if another shared Channel object close the underlying + // channel + if (state == GRPC_CHANNEL_SHUTDOWN) { + channel->wrapper->wrapped = NULL; + } + gpr_mu_unlock(&channel->wrapper->mu); + RETURN_LONG(state); } /** @@ -194,25 +373,37 @@ PHP_METHOD(Channel, getConnectivityState) { */ PHP_METHOD(Channel, watchConnectivityState) { wrapped_grpc_channel *channel = Z_WRAPPED_GRPC_CHANNEL_P(getThis()); + gpr_mu_lock(&channel->wrapper->mu); + if (channel->wrapper->wrapped == NULL) { + zend_throw_exception(spl_ce_RuntimeException, + "Channel already closed", 1 TSRMLS_CC); + gpr_mu_unlock(&channel->wrapper->mu); + return; + } + php_grpc_long last_state; zval *deadline_obj; /* "lO" == 1 long 1 object */ if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "lO", - &last_state, &deadline_obj, grpc_ce_timeval) == FAILURE) { + &last_state, &deadline_obj, + grpc_ce_timeval) == FAILURE) { zend_throw_exception(spl_ce_InvalidArgumentException, - "watchConnectivityState expects 1 long 1 timeval", 1 TSRMLS_CC); + "watchConnectivityState expects 1 long 1 timeval", + 1 TSRMLS_CC); + gpr_mu_unlock(&channel->wrapper->mu); return; } wrapped_grpc_timeval *deadline = Z_WRAPPED_GRPC_TIMEVAL_P(deadline_obj); - grpc_channel_watch_connectivity_state(channel->wrapped, + grpc_channel_watch_connectivity_state(channel->wrapper->wrapped, (grpc_connectivity_state)last_state, deadline->wrapped, completion_queue, NULL); grpc_event event = - grpc_completion_queue_pluck(completion_queue, NULL, - gpr_inf_future(GPR_CLOCK_REALTIME), NULL); + grpc_completion_queue_pluck(completion_queue, NULL, + gpr_inf_future(GPR_CLOCK_REALTIME), NULL); + gpr_mu_unlock(&channel->wrapper->mu); RETURN_BOOL(event.success); } @@ -222,10 +413,48 @@ PHP_METHOD(Channel, watchConnectivityState) { */ PHP_METHOD(Channel, close) { wrapped_grpc_channel *channel = Z_WRAPPED_GRPC_CHANNEL_P(getThis()); - if (channel->wrapped != NULL) { - grpc_channel_destroy(channel->wrapped); - channel->wrapped = NULL; + gpr_mu_lock(&channel->wrapper->mu); + if (channel->wrapper->wrapped != NULL) { + grpc_channel_destroy(channel->wrapper->wrapped); + channel->wrapper->wrapped = NULL; + } + + php_grpc_delete_persistent_list_entry(channel->wrapper->key, + strlen(channel->wrapper->key) + TSRMLS_CC); + gpr_mu_unlock(&channel->wrapper->mu); +} + +// Delete an entry from the persistent list +// Note: this does not destroy or close the underlying grpc_channel +void php_grpc_delete_persistent_list_entry(char *key, php_grpc_int key_len + TSRMLS_DC) { + php_grpc_zend_resource *rsrc; + gpr_mu_lock(&global_persistent_list_mu); + if (PHP_GRPC_PERSISTENT_LIST_FIND(&EG(persistent_list), key, + key_len, rsrc)) { + channel_persistent_le_t *le; + le = (channel_persistent_le_t *)rsrc->ptr; + le->channel = NULL; + php_grpc_zend_hash_del(&EG(persistent_list), key, key_len+1); + } + gpr_mu_unlock(&global_persistent_list_mu); +} + +// A destructor associated with each list entry from the persistent list +static void php_grpc_channel_plink_dtor(php_grpc_zend_resource *rsrc + TSRMLS_DC) { + channel_persistent_le_t *le = (channel_persistent_le_t *)rsrc->ptr; + if (le->channel != NULL) { + gpr_mu_lock(&le->channel->mu); + if (le->channel->wrapped != NULL) { + grpc_channel_destroy(le->channel->wrapped); + free(le->channel->key); + free(le->channel); + } + gpr_mu_unlock(&le->channel->mu); } + free(le); } ZEND_BEGIN_ARG_INFO_EX(arginfo_construct, 0, 0, 2) @@ -262,10 +491,13 @@ static zend_function_entry channel_methods[] = { PHP_FE_END }; -void grpc_init_channel(TSRMLS_D) { +GRPC_STARTUP_FUNCTION(channel) { zend_class_entry ce; INIT_CLASS_ENTRY(ce, "Grpc\\Channel", channel_methods); ce.create_object = create_wrapped_grpc_channel; grpc_ce_channel = zend_register_internal_class(&ce TSRMLS_CC); + le_plink = zend_register_list_destructors_ex( + NULL, php_grpc_channel_plink_dtor, "Persistent Channel", module_number); PHP_GRPC_INIT_HANDLER(wrapped_grpc_channel, channel_ce_handlers); + return SUCCESS; } diff --git a/src/php/ext/grpc/channel.h b/src/php/ext/grpc/channel.h index 45c9744135..69adc4782c 100755 --- a/src/php/ext/grpc/channel.h +++ b/src/php/ext/grpc/channel.h @@ -33,9 +33,18 @@ /* Class entry for the PHP Channel class */ extern zend_class_entry *grpc_ce_channel; +typedef struct _grpc_channel_wrapper { + grpc_channel *wrapped; + char *key; + char *target; + char *args_hashstr; + char *creds_hashstr; + gpr_mu mu; +} grpc_channel_wrapper; + /* Wrapper struct for grpc_channel that can be associated with a PHP object */ PHP_GRPC_WRAP_OBJECT_START(wrapped_grpc_channel) - grpc_channel *wrapped; + grpc_channel_wrapper *wrapper; PHP_GRPC_WRAP_OBJECT_END(wrapped_grpc_channel) #if PHP_MAJOR_VERSION < 7 @@ -57,10 +66,20 @@ static inline wrapped_grpc_channel #endif /* PHP_MAJOR_VERSION */ /* Initializes the Channel class */ -void grpc_init_channel(TSRMLS_D); +GRPC_STARTUP_FUNCTION(channel); /* Iterates through a PHP array and populates args with the contents */ -void php_grpc_read_args_array(zval *args_array, grpc_channel_args *args - TSRMLS_DC); +int php_grpc_read_args_array(zval *args_array, grpc_channel_args *args + TSRMLS_DC); + +void generate_sha1_str(char *sha1str, char *str, php_grpc_int len); + +void php_grpc_delete_persistent_list_entry(char *key, php_grpc_int key_len + TSRMLS_DC); + +typedef struct _channel_persistent_le { + grpc_channel_wrapper *channel; +} channel_persistent_le_t; + #endif /* NET_GRPC_PHP_GRPC_CHANNEL_H_ */ diff --git a/src/php/ext/grpc/channel_credentials.c b/src/php/ext/grpc/channel_credentials.c index 40629c8b00..19e1cefb6f 100644 --- a/src/php/ext/grpc/channel_credentials.c +++ b/src/php/ext/grpc/channel_credentials.c @@ -26,7 +26,9 @@ #include <php.h> #include <php_ini.h> #include <ext/standard/info.h> +#include <ext/standard/sha1.h> #include <ext/spl/spl_exceptions.h> +#include "channel.h" #include "php_grpc.h" #include <zend_exceptions.h> @@ -69,14 +71,17 @@ php_grpc_zend_object create_wrapped_grpc_channel_credentials( channel_credentials_ce_handlers); } -zval *grpc_php_wrap_channel_credentials(grpc_channel_credentials - *wrapped TSRMLS_DC) { +zval *grpc_php_wrap_channel_credentials(grpc_channel_credentials *wrapped, + char *hashstr, + zend_bool has_call_creds TSRMLS_DC) { zval *credentials_object; PHP_GRPC_MAKE_STD_ZVAL(credentials_object); object_init_ex(credentials_object, grpc_ce_channel_credentials); wrapped_grpc_channel_credentials *credentials = Z_WRAPPED_GRPC_CHANNEL_CREDS_P(credentials_object); credentials->wrapped = wrapped; + credentials->hashstr = hashstr; + credentials->has_call_creds = has_call_creds; return credentials_object; } @@ -106,7 +111,8 @@ PHP_METHOD(ChannelCredentials, setDefaultRootsPem) { */ PHP_METHOD(ChannelCredentials, createDefault) { grpc_channel_credentials *creds = grpc_google_default_credentials_create(); - zval *creds_object = grpc_php_wrap_channel_credentials(creds TSRMLS_CC); + zval *creds_object = grpc_php_wrap_channel_credentials(creds, NULL, false + TSRMLS_CC); RETURN_DESTROY_ZVAL(creds_object); } @@ -140,10 +146,24 @@ PHP_METHOD(ChannelCredentials, createSsl) { "createSsl expects 3 optional strings", 1 TSRMLS_CC); return; } + + php_grpc_int hashkey_len = root_certs_length + cert_chain_length; + char hashkey[hashkey_len]; + if (root_certs_length > 0) { + strcpy(hashkey, pem_root_certs); + } + if (cert_chain_length > 0) { + strcpy(hashkey, pem_key_cert_pair.cert_chain); + } + + char *hashstr = malloc(41); + generate_sha1_str(hashstr, hashkey, hashkey_len); + grpc_channel_credentials *creds = grpc_ssl_credentials_create( pem_root_certs, pem_key_cert_pair.private_key == NULL ? NULL : &pem_key_cert_pair, NULL); - zval *creds_object = grpc_php_wrap_channel_credentials(creds TSRMLS_CC); + zval *creds_object = grpc_php_wrap_channel_credentials(creds, hashstr, false + TSRMLS_CC); RETURN_DESTROY_ZVAL(creds_object); } @@ -172,7 +192,9 @@ PHP_METHOD(ChannelCredentials, createComposite) { grpc_channel_credentials *creds = grpc_composite_channel_credentials_create(cred1->wrapped, cred2->wrapped, NULL); - zval *creds_object = grpc_php_wrap_channel_credentials(creds TSRMLS_CC); + zval *creds_object = + grpc_php_wrap_channel_credentials(creds, cred1->hashstr, true + TSRMLS_CC); RETURN_DESTROY_ZVAL(creds_object); } diff --git a/src/php/ext/grpc/channel_credentials.h b/src/php/ext/grpc/channel_credentials.h index 28c7f2c1d3..357d732642 100755 --- a/src/php/ext/grpc/channel_credentials.h +++ b/src/php/ext/grpc/channel_credentials.h @@ -38,6 +38,8 @@ extern zend_class_entry *grpc_ce_channel_credentials; * with a PHP object */ PHP_GRPC_WRAP_OBJECT_START(wrapped_grpc_channel_credentials) grpc_channel_credentials *wrapped; + char *hashstr; + zend_bool has_call_creds; PHP_GRPC_WRAP_OBJECT_END(wrapped_grpc_channel_credentials) #if PHP_MAJOR_VERSION < 7 diff --git a/src/php/ext/grpc/php7_wrapper.h b/src/php/ext/grpc/php7_wrapper.h index d4b4c262a7..96091f9dad 100644 --- a/src/php/ext/grpc/php7_wrapper.h +++ b/src/php/ext/grpc/php7_wrapper.h @@ -113,6 +113,20 @@ static inline int php_grpc_zend_hash_find(HashTable *ht, char *key, int len, } #define php_grpc_zend_hash_del zend_hash_del +#define php_grpc_zend_resource zend_rsrc_list_entry + +#define PHP_GRPC_BVAL_IS_TRUE(zv) Z_LVAL_P(zv) +#define PHP_GRPC_VAR_SERIALIZE(buf, zv, hash) \ + php_var_serialize(buf, &zv, hash TSRMLS_CC) +#define PHP_GRPC_SERIALIZED_BUF_STR(buf) buf.c +#define PHP_GRPC_SERIALIZED_BUF_LEN(buf) buf.len +#define PHP_GRPC_SHA1Update(cxt, str, len) \ + PHP_SHA1Update(cxt, (const unsigned char *)str, len) +#define PHP_GRPC_PERSISTENT_LIST_FIND(plist, key, len, rsrc) \ + zend_hash_find(plist, key, len+1, (void **)&rsrc) != FAILURE +#define PHP_GRPC_PERSISTENT_LIST_UPDATE(plist, key, len, rsrc) \ + zend_hash_update(plist, key, len+1, rsrc, sizeof(php_grpc_zend_resource), \ + NULL) #define PHP_GRPC_GET_CLASS_ENTRY(object) zend_get_class_entry(object TSRMLS_CC) @@ -200,6 +214,20 @@ static inline int php_grpc_zend_hash_find(HashTable *ht, char *key, int len, static inline int php_grpc_zend_hash_del(HashTable *ht, char *key, int len) { return zend_hash_str_del(ht, key, len - 1); } +#define php_grpc_zend_resource zend_resource + +#define PHP_GRPC_BVAL_IS_TRUE(zv) Z_TYPE_P(zv) == IS_TRUE +#define PHP_GRPC_VAR_SERIALIZE(buf, zv, hash) \ + php_var_serialize(buf, zv, hash) +#define PHP_GRPC_SERIALIZED_BUF_STR(buf) ZSTR_VAL(buf.s) +#define PHP_GRPC_SERIALIZED_BUF_LEN(buf) ZSTR_LEN(buf.s) +#define PHP_GRPC_SHA1Update(cxt, str, len) \ + PHP_SHA1Update(cxt, (unsigned char *)str, len) +#define PHP_GRPC_PERSISTENT_LIST_FIND(plist, key, len, rsrc) \ + (rsrc = zend_hash_str_find_ptr(plist, key, len)) != NULL +#define PHP_GRPC_PERSISTENT_LIST_UPDATE(plist, key, len, rsrc) \ + zend_hash_str_update_mem(plist, key, len, rsrc, \ + sizeof(php_grpc_zend_resource)) #define PHP_GRPC_GET_CLASS_ENTRY(object) Z_OBJ_P(object)->ce diff --git a/src/php/ext/grpc/php_grpc.c b/src/php/ext/grpc/php_grpc.c index 281b9e6aba..a96daf7d9b 100644 --- a/src/php/ext/grpc/php_grpc.c +++ b/src/php/ext/grpc/php_grpc.c @@ -221,7 +221,7 @@ PHP_MINIT_FUNCTION(grpc) { CONST_CS | CONST_PERSISTENT); grpc_init_call(TSRMLS_C); - grpc_init_channel(TSRMLS_C); + GRPC_STARTUP(channel); grpc_init_server(TSRMLS_C); grpc_init_timeval(TSRMLS_C); grpc_init_channel_credentials(TSRMLS_C); diff --git a/src/php/ext/grpc/php_grpc.h b/src/php/ext/grpc/php_grpc.h index ed846fdba4..32329178dc 100644 --- a/src/php/ext/grpc/php_grpc.h +++ b/src/php/ext/grpc/php_grpc.h @@ -74,4 +74,8 @@ ZEND_END_MODULE_GLOBALS(grpc) #define GRPC_G(v) (grpc_globals.v) #endif +#define GRPC_STARTUP_FUNCTION(module) ZEND_MINIT_FUNCTION(grpc_##module) +#define GRPC_STARTUP(module) \ + ZEND_MODULE_STARTUP_N(grpc_##module)(INIT_FUNC_ARGS_PASSTHRU) + #endif /* PHP_GRPC_H */ diff --git a/src/php/tests/unit_tests/CallTest.php b/src/php/tests/unit_tests/CallTest.php index 3270e73f82..c5e1890a98 100644 --- a/src/php/tests/unit_tests/CallTest.php +++ b/src/php/tests/unit_tests/CallTest.php @@ -37,8 +37,7 @@ class CallTest extends PHPUnit_Framework_TestCase public function tearDown() { - unset($this->call); - unset($this->channel); + $this->channel->close(); } public function testConstructor() diff --git a/src/php/tests/unit_tests/ChannelTest.php b/src/php/tests/unit_tests/ChannelTest.php index 34e6185031..400df0fb66 100644 --- a/src/php/tests/unit_tests/ChannelTest.php +++ b/src/php/tests/unit_tests/ChannelTest.php @@ -25,17 +25,15 @@ class ChannelTest extends PHPUnit_Framework_TestCase public function tearDown() { - unset($this->channel); + if (!empty($this->channel)) { + $this->channel->close(); + } } public function testInsecureCredentials() { - $this->channel = new Grpc\Channel( - 'localhost:0', - [ - 'credentials' => Grpc\ChannelCredentials::createInsecure(), - ] - ); + $this->channel = new Grpc\Channel('localhost:0', + ['credentials' => Grpc\ChannelCredentials::createInsecure()]); $this->assertSame('Grpc\Channel', get_class($this->channel)); } @@ -111,7 +109,7 @@ class ChannelTest extends PHPUnit_Framework_TestCase */ public function testInvalidConstructorWith() { - $this->channel = new Grpc\Channel('localhost', 'invalid'); + $this->channel = new Grpc\Channel('localhost:0', 'invalid'); $this->assertNull($this->channel); } @@ -120,12 +118,8 @@ class ChannelTest extends PHPUnit_Framework_TestCase */ public function testInvalidCredentials() { - $this->channel = new Grpc\Channel( - 'localhost:0', - [ - 'credentials' => new Grpc\Timeval(100), - ] - ); + $this->channel = new Grpc\Channel('localhost:0', + ['credentials' => new Grpc\Timeval(100)]); } /** @@ -133,12 +127,8 @@ class ChannelTest extends PHPUnit_Framework_TestCase */ public function testInvalidOptionsArray() { - $this->channel = new Grpc\Channel( - 'localhost:0', - [ - 'abc' => [], - ] - ); + $this->channel = new Grpc\Channel('localhost:0', + ['abc' => []]); } /** @@ -170,4 +160,431 @@ class ChannelTest extends PHPUnit_Framework_TestCase ['credentials' => Grpc\ChannelCredentials::createInsecure()]); $this->channel->watchConnectivityState(1, 'hi'); } + + + public function assertConnecting($state) { + $this->assertTrue($state == GRPC\CHANNEL_CONNECTING || + $state == GRPC\CHANNEL_TRANSIENT_FAILURE); + } + + public function waitUntilNotIdle($channel) { + for ($i = 0; $i < 10; $i++) { + $now = Grpc\Timeval::now(); + $deadline = $now->add(new Grpc\Timeval(1000)); + if ($channel->watchConnectivityState(GRPC\CHANNEL_IDLE, + $deadline)) { + return true; + } + } + $this->assertTrue(false); + } + + public function testPersistentChannelSameHost() + { + $this->channel1 = new Grpc\Channel('localhost:1', []); + // the underlying grpc channel is the same by default + // when connecting to the same host + $this->channel2 = new Grpc\Channel('localhost:1', []); + + // both channels should be IDLE + $state = $this->channel1->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + // both channels should now be in the CONNECTING state + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertConnecting($state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelDifferentHost() + { + // two different underlying channels because different hostname + $this->channel1 = new Grpc\Channel('localhost:1', []); + $this->channel2 = new Grpc\Channel('localhost:2', []); + + // both channels should be IDLE + $state = $this->channel1->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + // channel1 should now be in the CONNECTING state + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + // channel2 should still be in the IDLE state + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelSameArgs() + { + $this->channel1 = new Grpc\Channel('localhost:1', ["abc" => "def"]); + $this->channel2 = new Grpc\Channel('localhost:1', ["abc" => "def"]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertConnecting($state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelDifferentArgs() + { + $this->channel1 = new Grpc\Channel('localhost:1', []); + $this->channel2 = new Grpc\Channel('localhost:1', ["abc" => "def"]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelSameChannelCredentials() + { + $creds1 = Grpc\ChannelCredentials::createSsl(); + $creds2 = Grpc\ChannelCredentials::createSsl(); + + $this->channel1 = new Grpc\Channel('localhost:1', + ["credentials" => $creds1]); + $this->channel2 = new Grpc\Channel('localhost:1', + ["credentials" => $creds2]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertConnecting($state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelDifferentChannelCredentials() + { + $creds1 = Grpc\ChannelCredentials::createSsl(); + $creds2 = Grpc\ChannelCredentials::createSsl( + file_get_contents(dirname(__FILE__).'/../data/ca.pem')); + + $this->channel1 = new Grpc\Channel('localhost:1', + ["credentials" => $creds1]); + $this->channel2 = new Grpc\Channel('localhost:1', + ["credentials" => $creds2]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelSameChannelCredentialsRootCerts() + { + $creds1 = Grpc\ChannelCredentials::createSsl( + file_get_contents(dirname(__FILE__).'/../data/ca.pem')); + $creds2 = Grpc\ChannelCredentials::createSsl( + file_get_contents(dirname(__FILE__).'/../data/ca.pem')); + + $this->channel1 = new Grpc\Channel('localhost:1', + ["credentials" => $creds1]); + $this->channel2 = new Grpc\Channel('localhost:1', + ["credentials" => $creds2]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertConnecting($state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelDifferentSecureChannelCredentials() + { + $creds1 = Grpc\ChannelCredentials::createSsl(); + $creds2 = Grpc\ChannelCredentials::createInsecure(); + + $this->channel1 = new Grpc\Channel('localhost:1', + ["credentials" => $creds1]); + $this->channel2 = new Grpc\Channel('localhost:1', + ["credentials" => $creds2]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel1->close(); + $this->channel2->close(); + } + + /** + * @expectedException RuntimeException + */ + public function testPersistentChannelSharedChannelClose() + { + // same underlying channel + $this->channel1 = new Grpc\Channel('localhost:1', []); + $this->channel2 = new Grpc\Channel('localhost:1', []); + + // close channel1 + $this->channel1->close(); + + // channel2 is now in SHUTDOWN state + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_FATAL_FAILURE, $state); + + // calling it again will result in an exception because the + // channel is already closed + $state = $this->channel2->getConnectivityState(); + } + + public function testPersistentChannelCreateAfterClose() + { + $this->channel1 = new Grpc\Channel('localhost:1', []); + + $this->channel1->close(); + + $this->channel2 = new Grpc\Channel('localhost:1', []); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel2->close(); + } + + public function testPersistentChannelSharedMoreThanTwo() + { + $this->channel1 = new Grpc\Channel('localhost:1', []); + $this->channel2 = new Grpc\Channel('localhost:1', []); + $this->channel3 = new Grpc\Channel('localhost:1', []); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + // all 3 channels should be in CONNECTING state + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel3->getConnectivityState(); + $this->assertConnecting($state); + + $this->channel1->close(); + } + + public function callbackFunc($context) + { + return []; + } + + public function callbackFunc2($context) + { + return ["k1" => "v1"]; + } + + public function testPersistentChannelWithCallCredentials() + { + $creds = Grpc\ChannelCredentials::createSsl(); + $callCreds = Grpc\CallCredentials::createFromPlugin( + [$this, 'callbackFunc']); + $credsWithCallCreds = Grpc\ChannelCredentials::createComposite( + $creds, $callCreds); + + // If a ChannelCredentials object is composed with a + // CallCredentials object, the underlying grpc channel will + // always be created new and NOT persisted. + $this->channel1 = new Grpc\Channel('localhost:1', + ["credentials" => + $credsWithCallCreds]); + $this->channel2 = new Grpc\Channel('localhost:1', + ["credentials" => + $credsWithCallCreds]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelWithDifferentCallCredentials() + { + $callCreds1 = Grpc\CallCredentials::createFromPlugin( + [$this, 'callbackFunc']); + $callCreds2 = Grpc\CallCredentials::createFromPlugin( + [$this, 'callbackFunc2']); + + $creds1 = Grpc\ChannelCredentials::createSsl(); + $creds2 = Grpc\ChannelCredentials::createComposite( + $creds1, $callCreds1); + $creds3 = Grpc\ChannelCredentials::createComposite( + $creds1, $callCreds2); + + // Similar to the test above, anytime a ChannelCredentials + // object is composed with a CallCredentials object, the + // underlying grpc channel will always be separate and not + // persisted + $this->channel1 = new Grpc\Channel('localhost:1', + ["credentials" => $creds1]); + $this->channel2 = new Grpc\Channel('localhost:1', + ["credentials" => $creds2]); + $this->channel3 = new Grpc\Channel('localhost:1', + ["credentials" => $creds3]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + $state = $this->channel3->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel1->close(); + $this->channel2->close(); + $this->channel3->close(); + } + + public function testPersistentChannelForceNew() + { + $this->channel1 = new Grpc\Channel('localhost:1', []); + // even though all the channel params are the same, channel2 + // has a new and different underlying channel + $this->channel2 = new Grpc\Channel('localhost:1', + ["force_new" => true]); + + // try to connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + // any dangling old connection to the same host must be + // manually closed + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelForceNewOldChannelIdle() + { + + $this->channel1 = new Grpc\Channel('localhost:1', []); + $this->channel2 = new Grpc\Channel('localhost:1', + ["force_new" => true]); + $this->channel3 = new Grpc\Channel('localhost:1', []); + + // try to connect on channel2 + $state = $this->channel2->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel2); + + $state = $this->channel1->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + $state = $this->channel2->getConnectivityState(); + $this->assertConnecting($state); + $state = $this->channel3->getConnectivityState(); + $this->assertConnecting($state); + + $this->channel1->close(); + $this->channel2->close(); + } + + public function testPersistentChannelForceNewOldChannelClose() + { + + $this->channel1 = new Grpc\Channel('localhost:1', []); + $this->channel2 = new Grpc\Channel('localhost:1', + ["force_new" => true]); + $this->channel3 = new Grpc\Channel('localhost:1', []); + + $this->channel1->close(); + + $state = $this->channel2->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + $state = $this->channel3->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + $this->channel2->close(); + $this->channel3->close(); + } + + public function testPersistentChannelForceNewNewChannelClose() + { + + $this->channel1 = new Grpc\Channel('localhost:1', []); + $this->channel2 = new Grpc\Channel('localhost:1', + ["force_new" => true]); + $this->channel3 = new Grpc\Channel('localhost:1', []); + + $this->channel2->close(); + + $state = $this->channel1->getConnectivityState(); + $this->assertEquals(GRPC\CHANNEL_IDLE, $state); + + // can still connect on channel1 + $state = $this->channel1->getConnectivityState(true); + $this->waitUntilNotIdle($this->channel1); + + $state = $this->channel1->getConnectivityState(); + $this->assertConnecting($state); + + $this->channel1->close(); + } } diff --git a/src/php/tests/unit_tests/EndToEndTest.php b/src/php/tests/unit_tests/EndToEndTest.php index 43d54d9ee6..b54f1d87c9 100644 --- a/src/php/tests/unit_tests/EndToEndTest.php +++ b/src/php/tests/unit_tests/EndToEndTest.php @@ -28,8 +28,7 @@ class EndToEndTest extends PHPUnit_Framework_TestCase public function tearDown() { - unset($this->channel); - unset($this->server); + $this->channel->close(); } public function testSimpleRequestBody() @@ -516,7 +515,7 @@ class EndToEndTest extends PHPUnit_Framework_TestCase $this->assertTrue($idle_state == Grpc\CHANNEL_IDLE); $now = Grpc\Timeval::now(); - $delta = new Grpc\Timeval(500000); // should timeout + $delta = new Grpc\Timeval(50000); // should timeout $deadline = $now->add($delta); $this->assertFalse($this->channel->watchConnectivityState( @@ -545,7 +544,7 @@ class EndToEndTest extends PHPUnit_Framework_TestCase $this->assertTrue($idle_state == Grpc\CHANNEL_IDLE); $now = Grpc\Timeval::now(); - $delta = new Grpc\Timeval(100000); + $delta = new Grpc\Timeval(50000); $deadline = $now->add($delta); $this->assertFalse($this->channel->watchConnectivityState( diff --git a/src/php/tests/unit_tests/SecureEndToEndTest.php b/src/php/tests/unit_tests/SecureEndToEndTest.php index 0fecbfb3dd..dff4e878ea 100644 --- a/src/php/tests/unit_tests/SecureEndToEndTest.php +++ b/src/php/tests/unit_tests/SecureEndToEndTest.php @@ -43,8 +43,7 @@ class SecureEndToEndTest extends PHPUnit_Framework_TestCase public function tearDown() { - unset($this->channel); - unset($this->server); + $this->channel->close(); } public function testSimpleRequestBody() diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index aa074df849..e52d43e81d 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -92,6 +92,9 @@ CORE_SOURCE_FILES = [ 'src/core/lib/iomgr/ev_windows.c', 'src/core/lib/iomgr/exec_ctx.c', 'src/core/lib/iomgr/executor.c', + 'src/core/lib/iomgr/gethostname_fallback.c', + 'src/core/lib/iomgr/gethostname_host_name_max.c', + 'src/core/lib/iomgr/gethostname_sysconf.c', 'src/core/lib/iomgr/iocp_windows.c', 'src/core/lib/iomgr/iomgr.c', 'src/core/lib/iomgr/iomgr_posix.c', diff --git a/src/python/grpcio_testing/grpc_testing/__init__.py b/src/python/grpcio_testing/grpc_testing/__init__.py index c5a17f457a..14e25f09e2 100644 --- a/src/python/grpcio_testing/grpc_testing/__init__.py +++ b/src/python/grpcio_testing/grpc_testing/__init__.py @@ -15,11 +15,284 @@ import abc +from google.protobuf import descriptor import six import grpc +class UnaryUnaryChannelRpc(six.with_metaclass(abc.ABCMeta)): + """Fixture for a unary-unary RPC invoked by a system under test. + + Enables users to "play server" for the RPC. + """ + + @abc.abstractmethod + def send_initial_metadata(self, initial_metadata): + """Sends the RPC's initial metadata to the system under test. + + Args: + initial_metadata: The RPC's initial metadata to be "sent" to + the system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def cancelled(self): + """Blocks until the system under test has cancelled the RPC.""" + raise NotImplementedError() + + @abc.abstractmethod + def terminate(self, response, trailing_metadata, code, details): + """Terminates the RPC. + + Args: + response: The response for the RPC. + trailing_metadata: The RPC's trailing metadata. + code: The RPC's status code. + details: The RPC's status details. + """ + raise NotImplementedError() + + +class UnaryStreamChannelRpc(six.with_metaclass(abc.ABCMeta)): + """Fixture for a unary-stream RPC invoked by a system under test. + + Enables users to "play server" for the RPC. + """ + + @abc.abstractmethod + def send_initial_metadata(self, initial_metadata): + """Sends the RPC's initial metadata to the system under test. + + Args: + initial_metadata: The RPC's initial metadata to be "sent" to + the system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def send_response(self, response): + """Sends a response to the system under test. + + Args: + response: A response message to be "sent" to the system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def cancelled(self): + """Blocks until the system under test has cancelled the RPC.""" + raise NotImplementedError() + + @abc.abstractmethod + def terminate(self, trailing_metadata, code, details): + """Terminates the RPC. + + Args: + trailing_metadata: The RPC's trailing metadata. + code: The RPC's status code. + details: The RPC's status details. + """ + raise NotImplementedError() + + +class StreamUnaryChannelRpc(six.with_metaclass(abc.ABCMeta)): + """Fixture for a stream-unary RPC invoked by a system under test. + + Enables users to "play server" for the RPC. + """ + + @abc.abstractmethod + def send_initial_metadata(self, initial_metadata): + """Sends the RPC's initial metadata to the system under test. + + Args: + initial_metadata: The RPC's initial metadata to be "sent" to + the system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def take_request(self): + """Draws one of the requests added to the RPC by the system under test. + + This method blocks until the system under test has added to the RPC + the request to be returned. + + Successive calls to this method return requests in the same order in + which the system under test added them to the RPC. + + Returns: + A request message added to the RPC by the system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def requests_closed(self): + """Blocks until the system under test has closed the request stream.""" + raise NotImplementedError() + + @abc.abstractmethod + def cancelled(self): + """Blocks until the system under test has cancelled the RPC.""" + raise NotImplementedError() + + @abc.abstractmethod + def terminate(self, response, trailing_metadata, code, details): + """Terminates the RPC. + + Args: + response: The response for the RPC. + trailing_metadata: The RPC's trailing metadata. + code: The RPC's status code. + details: The RPC's status details. + """ + raise NotImplementedError() + + +class StreamStreamChannelRpc(six.with_metaclass(abc.ABCMeta)): + """Fixture for a stream-stream RPC invoked by a system under test. + + Enables users to "play server" for the RPC. + """ + + @abc.abstractmethod + def send_initial_metadata(self, initial_metadata): + """Sends the RPC's initial metadata to the system under test. + + Args: + initial_metadata: The RPC's initial metadata to be "sent" to the + system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def take_request(self): + """Draws one of the requests added to the RPC by the system under test. + + This method blocks until the system under test has added to the RPC + the request to be returned. + + Successive calls to this method return requests in the same order in + which the system under test added them to the RPC. + + Returns: + A request message added to the RPC by the system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def send_response(self, response): + """Sends a response to the system under test. + + Args: + response: A response messages to be "sent" to the system under test. + """ + raise NotImplementedError() + + @abc.abstractmethod + def requests_closed(self): + """Blocks until the system under test has closed the request stream.""" + raise NotImplementedError() + + @abc.abstractmethod + def cancelled(self): + """Blocks until the system under test has cancelled the RPC.""" + raise NotImplementedError() + + @abc.abstractmethod + def terminate(self, trailing_metadata, code, details): + """Terminates the RPC. + + Args: + trailing_metadata: The RPC's trailing metadata. + code: The RPC's status code. + details: The RPC's status details. + """ + raise NotImplementedError() + + +class Channel(six.with_metaclass(abc.ABCMeta), grpc.Channel): + """A grpc.Channel double with which to test a system that invokes RPCs.""" + + @abc.abstractmethod + def take_unary_unary(self, method_descriptor): + """Draws an RPC currently being made by the system under test. + + If the given descriptor does not identify any RPC currently being made + by the system under test, this method blocks until the system under + test invokes such an RPC. + + Args: + method_descriptor: A descriptor.MethodDescriptor describing a + unary-unary RPC method. + + Returns: + A (invocation_metadata, request, unary_unary_channel_rpc) tuple of + the RPC's invocation metadata, its request, and a + UnaryUnaryChannelRpc with which to "play server" for the RPC. + """ + raise NotImplementedError() + + @abc.abstractmethod + def take_unary_stream(self, method_descriptor): + """Draws an RPC currently being made by the system under test. + + If the given descriptor does not identify any RPC currently being made + by the system under test, this method blocks until the system under + test invokes such an RPC. + + Args: + method_descriptor: A descriptor.MethodDescriptor describing a + unary-stream RPC method. + + Returns: + A (invocation_metadata, request, unary_stream_channel_rpc) tuple of + the RPC's invocation metadata, its request, and a + UnaryStreamChannelRpc with which to "play server" for the RPC. + """ + raise NotImplementedError() + + @abc.abstractmethod + def take_stream_unary(self, method_descriptor): + """Draws an RPC currently being made by the system under test. + + If the given descriptor does not identify any RPC currently being made + by the system under test, this method blocks until the system under + test invokes such an RPC. + + Args: + method_descriptor: A descriptor.MethodDescriptor describing a + stream-unary RPC method. + + Returns: + A (invocation_metadata, stream_unary_channel_rpc) tuple of the RPC's + invocation metadata and a StreamUnaryChannelRpc with which to "play + server" for the RPC. + """ + raise NotImplementedError() + + @abc.abstractmethod + def take_stream_stream(self, method_descriptor): + """Draws an RPC currently being made by the system under test. + + If the given descriptor does not identify any RPC currently being made + by the system under test, this method blocks until the system under + test invokes such an RPC. + + Args: + method_descriptor: A descriptor.MethodDescriptor describing a + stream-stream RPC method. + + Returns: + A (invocation_metadata, stream_stream_channel_rpc) tuple of the RPC's + invocation metadata and a StreamStreamChannelRpc with which to + "play server" for the RPC. + """ + raise NotImplementedError() + + class Time(six.with_metaclass(abc.ABCMeta)): """A simulation of time. @@ -117,3 +390,19 @@ def strict_fake_time(now): """ from grpc_testing import _time return _time.StrictFakeTime(now) + + +def channel(service_descriptors, time): + """Creates a Channel for use in tests of a gRPC Python-using system. + + Args: + service_descriptors: An iterable of descriptor.ServiceDescriptors + describing the RPCs that will be made on the returned Channel by the + system under test. + time: A Time to be used for tests. + + Returns: + A Channel for use in tests. + """ + from grpc_testing import _channel + return _channel.testing_channel(service_descriptors, time) diff --git a/src/python/grpcio_testing/grpc_testing/_channel/__init__.py b/src/python/grpcio_testing/grpc_testing/_channel/__init__.py new file mode 100644 index 0000000000..8011975d0a --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_channel/__init__.py @@ -0,0 +1,23 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from grpc_testing._channel import _channel +from grpc_testing._channel import _channel_state + + +# descriptors is reserved for later use. +# pylint: disable=unused-argument +def testing_channel(descriptors, time): + return _channel.TestingChannel(time, _channel_state.State()) +# pylint: enable=unused-argument diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_channel.py b/src/python/grpcio_testing/grpc_testing/_channel/_channel.py new file mode 100644 index 0000000000..fbd064db88 --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_channel/_channel.py @@ -0,0 +1,62 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import grpc_testing +from grpc_testing._channel import _channel_rpc +from grpc_testing._channel import _multi_callable + + +# All serializer and deserializer parameters are not (yet) used by this +# test infrastructure. +# pylint: disable=unused-argument +class TestingChannel(grpc_testing.Channel): + + def __init__(self, time, state): + self._time = time + self._state = state + + def subscribe(self, callback, try_to_connect=False): + raise NotImplementedError() + + def unsubscribe(self, callback): + raise NotImplementedError() + + def unary_unary( + self, method, request_serializer=None, response_deserializer=None): + return _multi_callable.UnaryUnary(method, self._state) + + def unary_stream( + self, method, request_serializer=None, response_deserializer=None): + return _multi_callable.UnaryStream(method, self._state) + + def stream_unary( + self, method, request_serializer=None, response_deserializer=None): + return _multi_callable.StreamUnary(method, self._state) + + def stream_stream( + self, method, request_serializer=None, response_deserializer=None): + return _multi_callable.StreamStream(method, self._state) + + def take_unary_unary(self, method_descriptor): + return _channel_rpc.unary_unary(self._state, method_descriptor) + + def take_unary_stream(self, method_descriptor): + return _channel_rpc.unary_stream(self._state, method_descriptor) + + def take_stream_unary(self, method_descriptor): + return _channel_rpc.stream_unary(self._state, method_descriptor) + + def take_stream_stream(self, method_descriptor): + return _channel_rpc.stream_stream(self._state, method_descriptor) +# pylint: enable=unused-argument diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_channel_rpc.py b/src/python/grpcio_testing/grpc_testing/_channel/_channel_rpc.py new file mode 100644 index 0000000000..762b6a035b --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_channel/_channel_rpc.py @@ -0,0 +1,119 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import grpc_testing + + +class _UnaryUnary(grpc_testing.UnaryUnaryChannelRpc): + + def __init__(self, rpc_state): + self._rpc_state = rpc_state + + def send_initial_metadata(self, initial_metadata): + self._rpc_state.send_initial_metadata(initial_metadata) + + def cancelled(self): + self._rpc_state.cancelled() + + def terminate(self, response, trailing_metadata, code, details): + self._rpc_state.terminate_with_response( + response, trailing_metadata, code, details) + + +class _UnaryStream(grpc_testing.UnaryStreamChannelRpc): + + def __init__(self, rpc_state): + self._rpc_state = rpc_state + + def send_initial_metadata(self, initial_metadata): + self._rpc_state.send_initial_metadata(initial_metadata) + + def send_response(self, response): + self._rpc_state.send_response(response) + + def cancelled(self): + self._rpc_state.cancelled() + + def terminate(self, trailing_metadata, code, details): + self._rpc_state.terminate(trailing_metadata, code, details) + + +class _StreamUnary(grpc_testing.StreamUnaryChannelRpc): + + def __init__(self, rpc_state): + self._rpc_state = rpc_state + + def send_initial_metadata(self, initial_metadata): + self._rpc_state.send_initial_metadata(initial_metadata) + + def take_request(self): + return self._rpc_state.take_request() + + def requests_closed(self): + return self._rpc_state.requests_closed() + + def cancelled(self): + self._rpc_state.cancelled() + + def terminate(self, response, trailing_metadata, code, details): + self._rpc_state.terminate_with_response( + response, trailing_metadata, code, details) + + +class _StreamStream(grpc_testing.StreamStreamChannelRpc): + + def __init__(self, rpc_state): + self._rpc_state = rpc_state + + def send_initial_metadata(self, initial_metadata): + self._rpc_state.send_initial_metadata(initial_metadata) + + def take_request(self): + return self._rpc_state.take_request() + + def send_response(self, response): + self._rpc_state.send_response(response) + + def requests_closed(self): + return self._rpc_state.requests_closed() + + def cancelled(self): + self._rpc_state.cancelled() + + def terminate(self, trailing_metadata, code, details): + self._rpc_state.terminate(trailing_metadata, code, details) + + +def unary_unary(channel_state, method_descriptor): + rpc_state = channel_state.take_rpc_state(method_descriptor) + invocation_metadata, request = ( + rpc_state.take_invocation_metadata_and_request()) + return invocation_metadata, request, _UnaryUnary(rpc_state) + + +def unary_stream(channel_state, method_descriptor): + rpc_state = channel_state.take_rpc_state(method_descriptor) + invocation_metadata, request = ( + rpc_state.take_invocation_metadata_and_request()) + return invocation_metadata, request, _UnaryStream(rpc_state) + + +def stream_unary(channel_state, method_descriptor): + rpc_state = channel_state.take_rpc_state(method_descriptor) + return rpc_state.take_invocation_metadata(), _StreamUnary(rpc_state) + + +def stream_stream(channel_state, method_descriptor): + rpc_state = channel_state.take_rpc_state(method_descriptor) + return rpc_state.take_invocation_metadata(), _StreamStream(rpc_state) diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_channel_state.py b/src/python/grpcio_testing/grpc_testing/_channel/_channel_state.py new file mode 100644 index 0000000000..569c41d79d --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_channel/_channel_state.py @@ -0,0 +1,48 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import collections +import threading + +from grpc_testing import _common +from grpc_testing._channel import _rpc_state + + +class State(_common.ChannelHandler): + + def __init__(self): + self._condition = threading.Condition() + self._rpc_states = collections.defaultdict(list) + + def invoke_rpc( + self, method_full_rpc_name, invocation_metadata, requests, + requests_closed, timeout): + rpc_state = _rpc_state.State( + invocation_metadata, requests, requests_closed) + with self._condition: + self._rpc_states[method_full_rpc_name].append(rpc_state) + self._condition.notify_all() + return rpc_state + + def take_rpc_state(self, method_descriptor): + method_full_rpc_name = '/{}/{}'.format( + method_descriptor.containing_service.full_name, + method_descriptor.name) + with self._condition: + while True: + method_rpc_states = self._rpc_states[method_full_rpc_name] + if method_rpc_states: + return method_rpc_states.pop(0) + else: + self._condition.wait() diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py b/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py new file mode 100644 index 0000000000..ebce652eeb --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py @@ -0,0 +1,322 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import threading + +import grpc + +_NOT_YET_OBSERVED = object() + + +def _cancel(handler): + return handler.cancel(grpc.StatusCode.CANCELLED, 'Locally cancelled!') + + +def _is_active(handler): + return handler.is_active() + + +def _time_remaining(unused_handler): + raise NotImplementedError() + + +def _add_callback(handler, callback): + return handler.add_callback(callback) + + +def _initial_metadata(handler): + return handler.initial_metadata() + + +def _trailing_metadata(handler): + trailing_metadata, unused_code, unused_details = handler.termination() + return trailing_metadata + + +def _code(handler): + unused_trailing_metadata, code, unused_details = handler.termination() + return code + + +def _details(handler): + unused_trailing_metadata, unused_code, details = handler.termination() + return details + + +class _Call(grpc.Call): + + def __init__(self, handler): + self._handler = handler + + def cancel(self): + _cancel(self._handler) + + def is_active(self): + return _is_active(self._handler) + + def time_remaining(self): + return _time_remaining(self._handler) + + def add_callback(self, callback): + return _add_callback(self._handler, callback) + + def initial_metadata(self): + return _initial_metadata(self._handler) + + def trailing_metadata(self): + return _trailing_metadata(self._handler) + + def code(self): + return _code(self._handler) + + def details(self): + return _details(self._handler) + + +class _RpcErrorCall(grpc.RpcError, grpc.Call): + + def __init__(self, handler): + self._handler = handler + + def cancel(self): + _cancel(self._handler) + + def is_active(self): + return _is_active(self._handler) + + def time_remaining(self): + return _time_remaining(self._handler) + + def add_callback(self, callback): + return _add_callback(self._handler, callback) + + def initial_metadata(self): + return _initial_metadata(self._handler) + + def trailing_metadata(self): + return _trailing_metadata(self._handler) + + def code(self): + return _code(self._handler) + + def details(self): + return _details(self._handler) + + +def _next(handler): + read = handler.take_response() + if read.code is None: + return read.response + elif read.code is grpc.StatusCode.OK: + raise StopIteration() + else: + raise _RpcErrorCall(handler) + + +class _HandlerExtras(object): + + def __init__(self): + self.condition = threading.Condition() + self.unary_response = _NOT_YET_OBSERVED + self.cancelled = False + + +def _with_extras_cancel(handler, extras): + with extras.condition: + if handler.cancel(grpc.StatusCode.CANCELLED, 'Locally cancelled!'): + extras.cancelled = True + return True + else: + return False + + +def _extras_without_cancelled(extras): + with extras.condition: + return extras.cancelled + + +def _running(handler): + return handler.is_active() + + +def _done(handler): + return not handler.is_active() + + +def _with_extras_unary_response(handler, extras): + with extras.condition: + if extras.unary_response is _NOT_YET_OBSERVED: + read = handler.take_response() + if read.code is None: + extras.unary_response = read.response + return read.response + else: + raise _RpcErrorCall(handler) + else: + return extras.unary_response + + +def _exception(unused_handler): + raise NotImplementedError('TODO!') + + +def _traceback(unused_handler): + raise NotImplementedError('TODO!') + + +def _add_done_callback(handler, callback, future): + adapted_callback = lambda: callback(future) + if not handler.add_callback(adapted_callback): + callback(future) + + +class _FutureCall(grpc.Future, grpc.Call): + + def __init__(self, handler, extras): + self._handler = handler + self._extras = extras + + def cancel(self): + return _with_extras_cancel(self._handler, self._extras) + + def cancelled(self): + return _extras_without_cancelled(self._extras) + + def running(self): + return _running(self._handler) + + def done(self): + return _done(self._handler) + + def result(self): + return _with_extras_unary_response(self._handler, self._extras) + + def exception(self): + return _exception(self._handler) + + def traceback(self): + return _traceback(self._handler) + + def add_done_callback(self, fn): + _add_done_callback(self._handler, fn, self) + + def is_active(self): + return _is_active(self._handler) + + def time_remaining(self): + return _time_remaining(self._handler) + + def add_callback(self, callback): + return _add_callback(self._handler, callback) + + def initial_metadata(self): + return _initial_metadata(self._handler) + + def trailing_metadata(self): + return _trailing_metadata(self._handler) + + def code(self): + return _code(self._handler) + + def details(self): + return _details(self._handler) + + +def consume_requests(request_iterator, handler): + + def _consume(): + while True: + try: + request = next(request_iterator) + added = handler.add_request(request) + if not added: + break + except StopIteration: + handler.close_requests() + break + except Exception: # pylint: disable=broad-except + details = 'Exception iterating requests!' + logging.exception(details) + handler.cancel(grpc.StatusCode.UNKNOWN, details) + + consumption = threading.Thread(target=_consume) + consumption.start() + + +def blocking_unary_response(handler): + read = handler.take_response() + if read.code is None: + unused_trailing_metadata, code, unused_details = handler.termination() + if code is grpc.StatusCode.OK: + return read.response + else: + raise _RpcErrorCall(handler) + else: + raise _RpcErrorCall(handler) + + +def blocking_unary_response_with_call(handler): + read = handler.take_response() + if read.code is None: + unused_trailing_metadata, code, unused_details = handler.termination() + if code is grpc.StatusCode.OK: + return read.response, _Call(handler) + else: + raise _RpcErrorCall(handler) + else: + raise _RpcErrorCall(handler) + + +def future_call(handler): + return _FutureCall(handler, _HandlerExtras()) + + +class ResponseIteratorCall(grpc.Call): + + def __init__(self, handler): + self._handler = handler + + def __iter__(self): + return self + + def __next__(self): + return _next(self._handler) + + def next(self): + return _next(self._handler) + + def cancel(self): + _cancel(self._handler) + + def is_active(self): + return _is_active(self._handler) + + def time_remaining(self): + return _time_remaining(self._handler) + + def add_callback(self, callback): + return _add_callback(self._handler, callback) + + def initial_metadata(self): + return _initial_metadata(self._handler) + + def trailing_metadata(self): + return _trailing_metadata(self._handler) + + def code(self): + return _code(self._handler) + + def details(self): + return _details(self._handler) diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_multi_callable.py b/src/python/grpcio_testing/grpc_testing/_channel/_multi_callable.py new file mode 100644 index 0000000000..fe69257f5b --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_channel/_multi_callable.py @@ -0,0 +1,115 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import grpc +from grpc_testing import _common +from grpc_testing._channel import _invocation + +# All per-call credentials parameters are unused by this test infrastructure. +# pylint: disable=unused-argument +class UnaryUnary(grpc.UnaryUnaryMultiCallable): + + def __init__(self, method_full_rpc_name, channel_handler): + self._method_full_rpc_name = method_full_rpc_name + self._channel_handler = channel_handler + + def __call__(self, request, timeout=None, metadata=None, credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, _common.fuss_with_metadata(metadata), + [request], True, timeout) + return _invocation.blocking_unary_response(rpc_handler) + + def with_call(self, request, timeout=None, metadata=None, credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, _common.fuss_with_metadata(metadata), + [request], True, timeout) + return _invocation.blocking_unary_response_with_call(rpc_handler) + + def future(self, request, timeout=None, metadata=None, credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, _common.fuss_with_metadata(metadata), + [request], True, timeout) + return _invocation.future_call(rpc_handler) + + +class UnaryStream(grpc.StreamStreamMultiCallable): + + def __init__(self, method_full_rpc_name, channel_handler): + self._method_full_rpc_name = method_full_rpc_name + self._channel_handler = channel_handler + + def __call__(self, request, timeout=None, metadata=None, credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [request], True, timeout) + return _invocation.ResponseIteratorCall(rpc_handler) + + +class StreamUnary(grpc.StreamUnaryMultiCallable): + + def __init__(self, method_full_rpc_name, channel_handler): + self._method_full_rpc_name = method_full_rpc_name + self._channel_handler = channel_handler + + def __call__(self, + request_iterator, + timeout=None, + metadata=None, + credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [], False, timeout) + _invocation.consume_requests(request_iterator, rpc_handler) + return _invocation.blocking_unary_response(rpc_handler) + + def with_call(self, + request_iterator, + timeout=None, + metadata=None, + credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [], False, timeout) + _invocation.consume_requests(request_iterator, rpc_handler) + return _invocation.blocking_unary_response_with_call(rpc_handler) + + def future(self, + request_iterator, + timeout=None, + metadata=None, + credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [], False, timeout) + _invocation.consume_requests(request_iterator, rpc_handler) + return _invocation.future_call(rpc_handler) + + +class StreamStream(grpc.StreamStreamMultiCallable): + + def __init__(self, method_full_rpc_name, channel_handler): + self._method_full_rpc_name = method_full_rpc_name + self._channel_handler = channel_handler + + def __call__(self, + request_iterator, + timeout=None, + metadata=None, + credentials=None): + rpc_handler = self._channel_handler.invoke_rpc( + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [], False, timeout) + _invocation.consume_requests(request_iterator, rpc_handler) + return _invocation.ResponseIteratorCall(rpc_handler) +# pylint: enable=unused-argument diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_rpc_state.py b/src/python/grpcio_testing/grpc_testing/_channel/_rpc_state.py new file mode 100644 index 0000000000..e1fa49a2a8 --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_channel/_rpc_state.py @@ -0,0 +1,193 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import threading + +import grpc +from grpc_testing import _common + + +class State(_common.ChannelRpcHandler): + + def __init__(self, invocation_metadata, requests, requests_closed): + self._condition = threading.Condition() + self._invocation_metadata = invocation_metadata + self._requests = requests + self._requests_closed = requests_closed + self._initial_metadata = None + self._responses = [] + self._trailing_metadata = None + self._code = None + self._details = None + + def initial_metadata(self): + with self._condition: + while True: + if self._initial_metadata is None: + if self._code is None: + self._condition.wait() + else: + return _common.FUSSED_EMPTY_METADATA + else: + return self._initial_metadata + + def add_request(self, request): + with self._condition: + if self._code is None and not self._requests_closed: + self._requests.append(request) + self._condition.notify_all() + return True + else: + return False + + def close_requests(self): + with self._condition: + if self._code is None and not self._requests_closed: + self._requests_closed = True + self._condition.notify_all() + + def take_response(self): + with self._condition: + while True: + if self._code is grpc.StatusCode.OK: + if self._responses: + response = self._responses.pop(0) + return _common.ChannelRpcRead( + response, None, None, None) + else: + return _common.ChannelRpcRead( + None, self._trailing_metadata, + grpc.StatusCode.OK, self._details) + elif self._code is None: + if self._responses: + response = self._responses.pop(0) + return _common.ChannelRpcRead( + response, None, None, None) + else: + self._condition.wait() + else: + return _common.ChannelRpcRead( + None, self._trailing_metadata, self._code, + self._details) + + def termination(self): + with self._condition: + while True: + if self._code is None: + self._condition.wait() + else: + return self._trailing_metadata, self._code, self._details + + def cancel(self, code, details): + with self._condition: + if self._code is None: + if self._initial_metadata is None: + self._initial_metadata = _common.FUSSED_EMPTY_METADATA + self._trailing_metadata = _common.FUSSED_EMPTY_METADATA + self._code = code + self._details = details + self._condition.notify_all() + return True + else: + return False + + def take_invocation_metadata(self): + with self._condition: + if self._invocation_metadata is None: + raise ValueError('Expected invocation metadata!') + else: + invocation_metadata = self._invocation_metadata + self._invocation_metadata = None + return invocation_metadata + + def take_invocation_metadata_and_request(self): + with self._condition: + if self._invocation_metadata is None: + raise ValueError('Expected invocation metadata!') + elif not self._requests: + raise ValueError('Expected at least one request!') + else: + invocation_metadata = self._invocation_metadata + self._invocation_metadata = None + return invocation_metadata, self._requests.pop(0) + + def send_initial_metadata(self, initial_metadata): + with self._condition: + self._initial_metadata = _common.fuss_with_metadata( + initial_metadata) + self._condition.notify_all() + + def take_request(self): + with self._condition: + while True: + if self._requests: + return self._requests.pop(0) + else: + self._condition.wait() + + def requests_closed(self): + with self._condition: + while True: + if self._requests_closed: + return + else: + self._condition.wait() + + def send_response(self, response): + with self._condition: + if self._code is None: + self._responses.append(response) + self._condition.notify_all() + + def terminate_with_response( + self, response, trailing_metadata, code, details): + with self._condition: + if self._initial_metadata is None: + self._initial_metadata = _common.FUSSED_EMPTY_METADATA + self._responses.append(response) + self._trailing_metadata = _common.fuss_with_metadata( + trailing_metadata) + self._code = code + self._details = details + self._condition.notify_all() + + def terminate(self, trailing_metadata, code, details): + with self._condition: + if self._initial_metadata is None: + self._initial_metadata = _common.FUSSED_EMPTY_METADATA + self._trailing_metadata = _common.fuss_with_metadata( + trailing_metadata) + self._code = code + self._details = details + self._condition.notify_all() + + def cancelled(self): + with self._condition: + while True: + if self._code is grpc.StatusCode.CANCELLED: + return + elif self._code is None: + self._condition.wait() + else: + raise ValueError( + 'Status code unexpectedly {}!'.format(self._code)) + + def is_active(self): + raise NotImplementedError() + + def time_remaining(self): + raise NotImplementedError() + + def add_callback(self, callback): + raise NotImplementedError() diff --git a/src/python/grpcio_testing/grpc_testing/_common.py b/src/python/grpcio_testing/grpc_testing/_common.py new file mode 100644 index 0000000000..cb4a7f5fa2 --- /dev/null +++ b/src/python/grpcio_testing/grpc_testing/_common.py @@ -0,0 +1,92 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Common interfaces and implementation.""" + +import abc +import collections + +import six + + +def _fuss(tuplified_metadata): + return tuplified_metadata + ( + ( + 'grpc.metadata_added_by_runtime', + 'gRPC is allowed to add metadata in transmission and does so.', + ), + ) + +FUSSED_EMPTY_METADATA = _fuss(()) + + +def fuss_with_metadata(metadata): + if metadata is None: + return FUSSED_EMPTY_METADATA + else: + return _fuss(tuple(metadata)) + + +class ChannelRpcRead( + collections.namedtuple( + 'ChannelRpcRead', + ('response', 'trailing_metadata', 'code', 'details',))): + pass + + +class ChannelRpcHandler(six.with_metaclass(abc.ABCMeta)): + + @abc.abstractmethod + def initial_metadata(self): + raise NotImplementedError() + + @abc.abstractmethod + def add_request(self, request): + raise NotImplementedError() + + @abc.abstractmethod + def close_requests(self): + raise NotImplementedError() + + @abc.abstractmethod + def take_response(self): + raise NotImplementedError() + + @abc.abstractmethod + def cancel(self, code, details): + raise NotImplementedError() + + @abc.abstractmethod + def termination(self): + raise NotImplementedError() + + @abc.abstractmethod + def is_active(self): + raise NotImplementedError() + + @abc.abstractmethod + def time_remaining(self): + raise NotImplementedError() + + @abc.abstractmethod + def add_callback(self, callback): + raise NotImplementedError() + + +class ChannelHandler(six.with_metaclass(abc.ABCMeta)): + + @abc.abstractmethod + def invoke_rpc( + self, method_full_rpc_name, invocation_metadata, requests, + requests_closed, timeout): + raise NotImplementedError() diff --git a/src/python/grpcio_tests/setup.py b/src/python/grpcio_tests/setup.py index adc909ccdc..debe14c40e 100644 --- a/src/python/grpcio_tests/setup.py +++ b/src/python/grpcio_tests/setup.py @@ -68,6 +68,10 @@ PACKAGE_DATA = { 'tests.protoc_plugin.protos.invocation_testing.split_services': [ 'services.proto', ], + 'tests.testing.proto': [ + 'requests.proto', + 'services.proto', + ], 'tests.unit': [ 'credentials/ca.pem', 'credentials/server1.key', diff --git a/src/python/grpcio_tests/tests/testing/_application_common.py b/src/python/grpcio_tests/tests/testing/_application_common.py new file mode 100644 index 0000000000..4e98879607 --- /dev/null +++ b/src/python/grpcio_tests/tests/testing/_application_common.py @@ -0,0 +1,36 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""An example gRPC Python-using application's common code elements.""" + +from tests.testing.proto import requests_pb2 +from tests.testing.proto import services_pb2 + +SERVICE_NAME = 'tests_of_grpc_testing.FirstService' +UNARY_UNARY_METHOD_NAME = 'UnUn' +UNARY_STREAM_METHOD_NAME = 'UnStre' +STREAM_UNARY_METHOD_NAME = 'StreUn' +STREAM_STREAM_METHOD_NAME = 'StreStre' + +UNARY_UNARY_REQUEST = requests_pb2.Up(first_up_field=2) +ERRONEOUS_UNARY_UNARY_REQUEST = requests_pb2.Up(first_up_field=3) +UNARY_UNARY_RESPONSE = services_pb2.Down(first_down_field=5) +ERRONEOUS_UNARY_UNARY_RESPONSE = services_pb2.Down(first_down_field=7) +UNARY_STREAM_REQUEST = requests_pb2.Charm(first_charm_field=11) +STREAM_UNARY_REQUEST = requests_pb2.Charm(first_charm_field=13) +STREAM_UNARY_RESPONSE = services_pb2.Strange(first_strange_field=17) +STREAM_STREAM_REQUEST = requests_pb2.Top(first_top_field=19) +STREAM_STREAM_RESPONSE = services_pb2.Bottom(first_bottom_field=23) +TWO_STREAM_STREAM_RESPONSES = (STREAM_STREAM_RESPONSE,) * 2 + +INFINITE_REQUEST_STREAM_TIMEOUT = 0.2 diff --git a/src/python/grpcio_tests/tests/testing/_application_testing_common.py b/src/python/grpcio_tests/tests/testing/_application_testing_common.py new file mode 100644 index 0000000000..9c9e485a78 --- /dev/null +++ b/src/python/grpcio_tests/tests/testing/_application_testing_common.py @@ -0,0 +1,33 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import grpc_testing + +from tests.testing.proto import requests_pb2 +from tests.testing.proto import services_pb2 + +# TODO(https://github.com/grpc/grpc/issues/11657): Eliminate this entirely. +# TODO(https://github.com/google/protobuf/issues/3452): Eliminate this if/else. +if services_pb2.DESCRIPTOR.services_by_name.get('FirstService') is None: + FIRST_SERVICE = 'Fix protobuf issue 3452!' + FIRST_SERVICE_UNUN = 'Fix protobuf issue 3452!' + FIRST_SERVICE_UNSTRE = 'Fix protobuf issue 3452!' + FIRST_SERVICE_STREUN = 'Fix protobuf issue 3452!' + FIRST_SERVICE_STRESTRE = 'Fix protobuf issue 3452!' +else: + FIRST_SERVICE = services_pb2.DESCRIPTOR.services_by_name['FirstService'] + FIRST_SERVICE_UNUN = FIRST_SERVICE.methods_by_name['UnUn'] + FIRST_SERVICE_UNSTRE = FIRST_SERVICE.methods_by_name['UnStre'] + FIRST_SERVICE_STREUN = FIRST_SERVICE.methods_by_name['StreUn'] + FIRST_SERVICE_STRESTRE = FIRST_SERVICE.methods_by_name['StreStre'] diff --git a/src/python/grpcio_tests/tests/testing/_client_application.py b/src/python/grpcio_tests/tests/testing/_client_application.py new file mode 100644 index 0000000000..aff32fb4dc --- /dev/null +++ b/src/python/grpcio_tests/tests/testing/_client_application.py @@ -0,0 +1,260 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""An example gRPC Python-using client-side application.""" + +import collections +import enum +import threading +import time + +import grpc +from tests.unit.framework.common import test_constants + +from tests.testing.proto import requests_pb2 +from tests.testing.proto import services_pb2 +from tests.testing.proto import services_pb2_grpc + +from tests.testing import _application_common + + +@enum.unique +class Scenario(enum.Enum): + UNARY_UNARY = 'unary unary' + UNARY_STREAM = 'unary stream' + STREAM_UNARY = 'stream unary' + STREAM_STREAM = 'stream stream' + CONCURRENT_STREAM_UNARY = 'concurrent stream unary' + CONCURRENT_STREAM_STREAM = 'concurrent stream stream' + CANCEL_UNARY_UNARY = 'cancel unary unary' + CANCEL_UNARY_STREAM = 'cancel unary stream' + INFINITE_REQUEST_STREAM = 'infinite request stream' + + +class Outcome(collections.namedtuple('Outcome', ('kind', 'code', 'details'))): + """Outcome of a client application scenario. + + Attributes: + kind: A Kind value describing the overall kind of scenario execution. + code: A grpc.StatusCode value. Only valid if kind is Kind.RPC_ERROR. + details: A status details string. Only valid if kind is Kind.RPC_ERROR. + """ + + @enum.unique + class Kind(enum.Enum): + SATISFACTORY = 'satisfactory' + UNSATISFACTORY = 'unsatisfactory' + RPC_ERROR = 'rpc error' + + +_SATISFACTORY_OUTCOME = Outcome(Outcome.Kind.SATISFACTORY, None, None) +_UNSATISFACTORY_OUTCOME = Outcome(Outcome.Kind.UNSATISFACTORY, None, None) + + +class _Pipe(object): + + def __init__(self): + self._condition = threading.Condition() + self._values = [] + self._open = True + + def __iter__(self): + return self + + def _next(self): + with self._condition: + while True: + if self._values: + return self._values.pop(0) + elif not self._open: + raise StopIteration() + else: + self._condition.wait() + + def __next__(self): # (Python 3 Iterator Protocol) + return self._next() + + def next(self): # (Python 2 Iterator Protocol) + return self._next() + + def add(self, value): + with self._condition: + self._values.append(value) + self._condition.notify_all() + + def close(self): + with self._condition: + self._open = False + self._condition.notify_all() + + +def _run_unary_unary(stub): + response = stub.UnUn(_application_common.UNARY_UNARY_REQUEST) + if _application_common.UNARY_UNARY_RESPONSE == response: + return _SATISFACTORY_OUTCOME + else: + return _UNSATISFACTORY_OUTCOME + + +def _run_unary_stream(stub): + response_iterator = stub.UnStre(_application_common.UNARY_STREAM_REQUEST) + try: + next(response_iterator) + except StopIteration: + return _SATISFACTORY_OUTCOME + else: + return _UNSATISFACTORY_OUTCOME + + +def _run_stream_unary(stub): + response, call = stub.StreUn.with_call( + iter((_application_common.STREAM_UNARY_REQUEST,) * 3)) + if (_application_common.STREAM_UNARY_RESPONSE == response and + call.code() is grpc.StatusCode.OK): + return _SATISFACTORY_OUTCOME + else: + return _UNSATISFACTORY_OUTCOME + + +def _run_stream_stream(stub): + request_pipe = _Pipe() + response_iterator = stub.StreStre(iter(request_pipe)) + request_pipe.add(_application_common.STREAM_STREAM_REQUEST) + first_responses = next(response_iterator), next(response_iterator), + request_pipe.add(_application_common.STREAM_STREAM_REQUEST) + second_responses = next(response_iterator), next(response_iterator), + request_pipe.close() + try: + next(response_iterator) + except StopIteration: + unexpected_extra_response = False + else: + unexpected_extra_response = True + if (first_responses == _application_common.TWO_STREAM_STREAM_RESPONSES and + second_responses == _application_common.TWO_STREAM_STREAM_RESPONSES + and not unexpected_extra_response): + return _SATISFACTORY_OUTCOME + else: + return _UNSATISFACTORY_OUTCOME + + +def _run_concurrent_stream_unary(stub): + future_calls = tuple( + stub.StreUn.future( + iter((_application_common.STREAM_UNARY_REQUEST,) * 3)) + for _ in range(test_constants.THREAD_CONCURRENCY)) + for future_call in future_calls: + if future_call.code() is grpc.StatusCode.OK: + response = future_call.result() + if _application_common.STREAM_UNARY_RESPONSE != response: + return _UNSATISFACTORY_OUTCOME + else: + return _UNSATISFACTORY_OUTCOME + else: + return _SATISFACTORY_OUTCOME + + +def _run_concurrent_stream_stream(stub): + condition = threading.Condition() + outcomes = [None] * test_constants.RPC_CONCURRENCY + + def run_stream_stream(index): + outcome = _run_stream_stream(stub) + with condition: + outcomes[index] = outcome + condition.notify() + + for index in range(test_constants.RPC_CONCURRENCY): + thread = threading.Thread(target=run_stream_stream, args=(index,)) + thread.start() + with condition: + while True: + if all(outcomes): + for outcome in outcomes: + if outcome.kind is not Outcome.Kind.SATISFACTORY: + return _UNSATISFACTORY_OUTCOME + else: + return _SATISFACTORY_OUTCOME + else: + condition.wait() + + +def _run_cancel_unary_unary(stub): + response_future_call = stub.UnUn.future( + _application_common.UNARY_UNARY_REQUEST) + initial_metadata = response_future_call.initial_metadata() + cancelled = response_future_call.cancel() + if initial_metadata is not None and cancelled: + return _SATISFACTORY_OUTCOME + else: + return _UNSATISFACTORY_OUTCOME + + +def _run_infinite_request_stream(stub): + + def infinite_request_iterator(): + while True: + yield _application_common.STREAM_UNARY_REQUEST + + response_future_call = stub.StreUn.future( + infinite_request_iterator(), + timeout=_application_common.INFINITE_REQUEST_STREAM_TIMEOUT) + if response_future_call.code() is grpc.StatusCode.DEADLINE_EXCEEDED: + return _SATISFACTORY_OUTCOME + else: + return _UNSATISFACTORY_OUTCOME + + +def run(scenario, channel): + stub = services_pb2_grpc.FirstServiceStub(channel) + try: + if scenario is Scenario.UNARY_UNARY: + return _run_unary_unary(stub) + elif scenario is Scenario.UNARY_STREAM: + return _run_unary_stream(stub) + elif scenario is Scenario.STREAM_UNARY: + return _run_stream_unary(stub) + elif scenario is Scenario.STREAM_STREAM: + return _run_stream_stream(stub) + elif scenario is Scenario.CONCURRENT_STREAM_UNARY: + return _run_concurrent_stream_unary(stub) + elif scenario is Scenario.CONCURRENT_STREAM_STREAM: + return _run_concurrent_stream_stream(stub) + elif scenario is Scenario.CANCEL_UNARY_UNARY: + return _run_cancel_unary_unary(stub) + elif scenario is Scenario.INFINITE_REQUEST_STREAM: + return _run_infinite_request_stream(stub) + except grpc.RpcError as rpc_error: + return Outcome(Outcome.Kind.RPC_ERROR, + rpc_error.code(), rpc_error.details()) + + +_IMPLEMENTATIONS = { + Scenario.UNARY_UNARY: _run_unary_unary, + Scenario.UNARY_STREAM: _run_unary_stream, + Scenario.STREAM_UNARY: _run_stream_unary, + Scenario.STREAM_STREAM: _run_stream_stream, + Scenario.CONCURRENT_STREAM_UNARY: _run_concurrent_stream_unary, + Scenario.CONCURRENT_STREAM_STREAM: _run_concurrent_stream_stream, + Scenario.CANCEL_UNARY_UNARY: _run_cancel_unary_unary, + Scenario.INFINITE_REQUEST_STREAM: _run_infinite_request_stream, +} + + +def run(scenario, channel): + stub = services_pb2_grpc.FirstServiceStub(channel) + try: + return _IMPLEMENTATIONS[scenario](stub) + except grpc.RpcError as rpc_error: + return Outcome(Outcome.Kind.RPC_ERROR, + rpc_error.code(), rpc_error.details()) diff --git a/src/python/grpcio_tests/tests/testing/_client_test.py b/src/python/grpcio_tests/tests/testing/_client_test.py new file mode 100644 index 0000000000..172f386d7b --- /dev/null +++ b/src/python/grpcio_tests/tests/testing/_client_test.py @@ -0,0 +1,306 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from concurrent import futures +import time +import unittest + +import grpc +from grpc.framework.foundation import logging_pool +from tests.unit.framework.common import test_constants +import grpc_testing + +from tests.testing import _application_common +from tests.testing import _application_testing_common +from tests.testing import _client_application +from tests.testing.proto import requests_pb2 +from tests.testing.proto import services_pb2 + + +# TODO(https://github.com/google/protobuf/issues/3452): Drop this skip. +@unittest.skipIf( + services_pb2.DESCRIPTOR.services_by_name.get('FirstService') is None, + 'Fix protobuf issue 3452!') +class ClientTest(unittest.TestCase): + + def setUp(self): + # In this test the client-side application under test executes in + # a separate thread while we retain use of the test thread to "play + # server". + self._client_execution_thread_pool = logging_pool.pool(1) + + self._fake_time = grpc_testing.strict_fake_time(time.time()) + self._real_time = grpc_testing.strict_real_time() + self._fake_time_channel = grpc_testing.channel( + services_pb2.DESCRIPTOR.services_by_name.values(), self._fake_time) + self._real_time_channel = grpc_testing.channel( + services_pb2.DESCRIPTOR.services_by_name.values(), self._real_time) + + def tearDown(self): + self._client_execution_thread_pool.shutdown(wait=True) + + def test_successful_unary_unary(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, _client_application.Scenario.UNARY_UNARY, + self._real_time_channel) + invocation_metadata, request, rpc = ( + self._real_time_channel.take_unary_unary( + _application_testing_common.FIRST_SERVICE_UNUN)) + rpc.send_initial_metadata(()) + rpc.terminate(_application_common.UNARY_UNARY_RESPONSE, (), + grpc.StatusCode.OK, '') + application_return_value = application_future.result() + + self.assertEqual(_application_common.UNARY_UNARY_REQUEST, request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.SATISFACTORY) + + def test_successful_unary_stream(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, _client_application.Scenario.UNARY_STREAM, + self._fake_time_channel) + invocation_metadata, request, rpc = ( + self._fake_time_channel.take_unary_stream( + _application_testing_common.FIRST_SERVICE_UNSTRE)) + rpc.send_initial_metadata(()) + rpc.terminate((), grpc.StatusCode.OK, '') + application_return_value = application_future.result() + + self.assertEqual(_application_common.UNARY_STREAM_REQUEST, request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.SATISFACTORY) + + def test_successful_stream_unary(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, _client_application.Scenario.STREAM_UNARY, + self._real_time_channel) + invocation_metadata, rpc = self._real_time_channel.take_stream_unary( + _application_testing_common.FIRST_SERVICE_STREUN) + rpc.send_initial_metadata(()) + first_request = rpc.take_request() + second_request = rpc.take_request() + third_request = rpc.take_request() + rpc.requests_closed() + rpc.terminate(_application_common.STREAM_UNARY_RESPONSE, (), + grpc.StatusCode.OK, '') + application_return_value = application_future.result() + + self.assertEqual(_application_common.STREAM_UNARY_REQUEST, + first_request) + self.assertEqual(_application_common.STREAM_UNARY_REQUEST, + second_request) + self.assertEqual(_application_common.STREAM_UNARY_REQUEST, + third_request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.SATISFACTORY) + + def test_successful_stream_stream(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, _client_application.Scenario.STREAM_STREAM, + self._fake_time_channel) + invocation_metadata, rpc = self._fake_time_channel.take_stream_stream( + _application_testing_common.FIRST_SERVICE_STRESTRE) + first_request = rpc.take_request() + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + second_request = rpc.take_request() + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.requests_closed() + rpc.terminate((), grpc.StatusCode.OK, '') + application_return_value = application_future.result() + + self.assertEqual(_application_common.STREAM_STREAM_REQUEST, + first_request) + self.assertEqual(_application_common.STREAM_STREAM_REQUEST, + second_request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.SATISFACTORY) + + def test_concurrent_stream_stream(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, + _client_application.Scenario.CONCURRENT_STREAM_STREAM, + self._real_time_channel) + rpcs = [] + for _ in range(test_constants.RPC_CONCURRENCY): + invocation_metadata, rpc = ( + self._real_time_channel.take_stream_stream( + _application_testing_common.FIRST_SERVICE_STRESTRE)) + rpcs.append(rpc) + requests = {} + for rpc in rpcs: + requests[rpc] = [rpc.take_request()] + for rpc in rpcs: + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + for rpc in rpcs: + requests[rpc].append(rpc.take_request()) + for rpc in rpcs: + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + for rpc in rpcs: + rpc.requests_closed() + for rpc in rpcs: + rpc.terminate((), grpc.StatusCode.OK, '') + application_return_value = application_future.result() + + for requests_of_one_rpc in requests.values(): + for request in requests_of_one_rpc: + self.assertEqual(_application_common.STREAM_STREAM_REQUEST, + request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.SATISFACTORY) + + def test_cancelled_unary_unary(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, + _client_application.Scenario.CANCEL_UNARY_UNARY, + self._fake_time_channel) + invocation_metadata, request, rpc = ( + self._fake_time_channel.take_unary_unary( + _application_testing_common.FIRST_SERVICE_UNUN)) + rpc.send_initial_metadata(()) + rpc.cancelled() + application_return_value = application_future.result() + + self.assertEqual(_application_common.UNARY_UNARY_REQUEST, request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.SATISFACTORY) + + def test_status_stream_unary(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, + _client_application.Scenario.CONCURRENT_STREAM_UNARY, + self._fake_time_channel) + rpcs = tuple( + self._fake_time_channel.take_stream_unary( + _application_testing_common.FIRST_SERVICE_STREUN)[1] + for _ in range(test_constants.THREAD_CONCURRENCY)) + for rpc in rpcs: + rpc.take_request() + rpc.take_request() + rpc.take_request() + rpc.requests_closed() + rpc.send_initial_metadata(( + ('my_metadata_key', 'My Metadata Value!',),)) + for rpc in rpcs[:-1]: + rpc.terminate(_application_common.STREAM_UNARY_RESPONSE, (), + grpc.StatusCode.OK, '') + rpcs[-1].terminate(_application_common.STREAM_UNARY_RESPONSE, (), + grpc.StatusCode.RESOURCE_EXHAUSTED, + 'nope; not able to handle all those RPCs!') + application_return_value = application_future.result() + + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.UNSATISFACTORY) + + def test_status_stream_stream(self): + code = grpc.StatusCode.DEADLINE_EXCEEDED + details = 'test deadline exceeded!' + + application_future = self._client_execution_thread_pool.submit( + _client_application.run, _client_application.Scenario.STREAM_STREAM, + self._real_time_channel) + invocation_metadata, rpc = self._real_time_channel.take_stream_stream( + _application_testing_common.FIRST_SERVICE_STRESTRE) + first_request = rpc.take_request() + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + second_request = rpc.take_request() + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.requests_closed() + rpc.terminate((), code, details) + application_return_value = application_future.result() + + self.assertEqual(_application_common.STREAM_STREAM_REQUEST, + first_request) + self.assertEqual(_application_common.STREAM_STREAM_REQUEST, + second_request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.RPC_ERROR) + self.assertIs(application_return_value.code, code) + self.assertEqual(application_return_value.details, details) + + def test_misbehaving_server_unary_unary(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, _client_application.Scenario.UNARY_UNARY, + self._fake_time_channel) + invocation_metadata, request, rpc = ( + self._fake_time_channel.take_unary_unary( + _application_testing_common.FIRST_SERVICE_UNUN)) + rpc.send_initial_metadata(()) + rpc.terminate(_application_common.ERRONEOUS_UNARY_UNARY_RESPONSE, (), + grpc.StatusCode.OK, '') + application_return_value = application_future.result() + + self.assertEqual(_application_common.UNARY_UNARY_REQUEST, request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.UNSATISFACTORY) + + def test_misbehaving_server_stream_stream(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, _client_application.Scenario.STREAM_STREAM, + self._real_time_channel) + invocation_metadata, rpc = self._real_time_channel.take_stream_stream( + _application_testing_common.FIRST_SERVICE_STRESTRE) + first_request = rpc.take_request() + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + second_request = rpc.take_request() + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.send_response(_application_common.STREAM_STREAM_RESPONSE) + rpc.requests_closed() + rpc.terminate((), grpc.StatusCode.OK, '') + application_return_value = application_future.result() + + self.assertEqual(_application_common.STREAM_STREAM_REQUEST, + first_request) + self.assertEqual(_application_common.STREAM_STREAM_REQUEST, + second_request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.UNSATISFACTORY) + + def test_infinite_request_stream_real_time(self): + application_future = self._client_execution_thread_pool.submit( + _client_application.run, + _client_application.Scenario.INFINITE_REQUEST_STREAM, + self._real_time_channel) + invocation_metadata, rpc = self._real_time_channel.take_stream_unary( + _application_testing_common.FIRST_SERVICE_STREUN) + rpc.send_initial_metadata(()) + first_request = rpc.take_request() + second_request = rpc.take_request() + third_request = rpc.take_request() + self._real_time.sleep_for( + _application_common.INFINITE_REQUEST_STREAM_TIMEOUT) + rpc.terminate(_application_common.STREAM_UNARY_RESPONSE, (), + grpc.StatusCode.DEADLINE_EXCEEDED, '') + application_return_value = application_future.result() + + self.assertEqual(_application_common.STREAM_UNARY_REQUEST, + first_request) + self.assertEqual(_application_common.STREAM_UNARY_REQUEST, + second_request) + self.assertEqual(_application_common.STREAM_UNARY_REQUEST, + third_request) + self.assertIs(application_return_value.kind, + _client_application.Outcome.Kind.SATISFACTORY) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/testing/proto/__init__.py b/src/python/grpcio_tests/tests/testing/proto/__init__.py new file mode 100644 index 0000000000..1e120359cf --- /dev/null +++ b/src/python/grpcio_tests/tests/testing/proto/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/src/python/grpcio_tests/tests/testing/proto/requests.proto b/src/python/grpcio_tests/tests/testing/proto/requests.proto new file mode 100644 index 0000000000..54a60bff86 --- /dev/null +++ b/src/python/grpcio_tests/tests/testing/proto/requests.proto @@ -0,0 +1,29 @@ +// Copyright 2015 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package tests_of_grpc_testing; + +message Up { + int32 first_up_field = 1; +} + +message Charm { + int32 first_charm_field = 1; +} + +message Top { + int32 first_top_field = 1; +} diff --git a/src/python/grpcio_tests/tests/testing/proto/services.proto b/src/python/grpcio_tests/tests/testing/proto/services.proto new file mode 100644 index 0000000000..cb15c0d1ce --- /dev/null +++ b/src/python/grpcio_tests/tests/testing/proto/services.proto @@ -0,0 +1,42 @@ +// Copyright 2017 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +import "tests/testing/proto/requests.proto"; + +package tests_of_grpc_testing; + +message Down { + int32 first_down_field = 1; +} + +message Strange { + int32 first_strange_field = 1; +} + +message Bottom { + int32 first_bottom_field = 1; +} + +service FirstService { + rpc UnUn(Up) returns (Down); + rpc UnStre(Charm) returns (stream Strange); + rpc StreUn(stream Charm) returns (Strange); + rpc StreStre(stream Top) returns (stream Bottom); +} + +service SecondService { + rpc UnStre(Strange) returns (stream Charm); +} diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json index f86eeb76c7..c10719b86f 100644 --- a/src/python/grpcio_tests/tests/tests.json +++ b/src/python/grpcio_tests/tests/tests.json @@ -9,6 +9,7 @@ "protoc_plugin._split_definitions_test.SplitSeparateTest", "protoc_plugin.beta_python_plugin_test.PythonPluginTest", "reflection._reflection_servicer_test.ReflectionServicerTest", + "testing._client_test.ClientTest", "testing._time_test.StrictFakeTimeTest", "testing._time_test.StrictRealTimeTest", "unit._api_test.AllTest", |