diff options
Diffstat (limited to 'src')
44 files changed, 294 insertions, 191 deletions
diff --git a/src/compiler/php_generator.cc b/src/compiler/php_generator.cc index 5dac02cec4..fba8cbaa97 100644 --- a/src/compiler/php_generator.cc +++ b/src/compiler/php_generator.cc @@ -134,29 +134,15 @@ void PrintService(const ServiceDescriptor *service, Printer *out) { out->Outdent(); out->Print("}\n\n"); } - -void PrintServices(const FileDescriptor *file, Printer *out) { - map<grpc::string, grpc::string> vars; - vars["package"] = MessageIdentifierName(file->package()); - out->Print(vars, "namespace $package$ {\n\n"); - out->Indent(); - for (int i = 0; i < file->service_count(); i++) { - PrintService(file->service(i), out); - } - out->Outdent(); - out->Print("}\n"); -} } -grpc::string GenerateFile(const FileDescriptor *file) { +grpc::string GenerateFile(const FileDescriptor *file, + const ServiceDescriptor *service) { grpc::string output; { StringOutputStream output_stream(&output); Printer out(&output_stream, '$'); - if (file->service_count() == 0) { - return output; - } out.Print("<?php\n"); out.Print("// GENERATED CODE -- DO NOT EDIT!\n\n"); @@ -166,7 +152,15 @@ grpc::string GenerateFile(const FileDescriptor *file) { out.Print(leading_comments.c_str()); } - PrintServices(file, &out); + map<grpc::string, grpc::string> vars; + vars["package"] = MessageIdentifierName(file->package()); + out.Print(vars, "namespace $package$ {\n\n"); + out.Indent(); + + PrintService(service, &out); + + out.Outdent(); + out.Print("}\n"); } return output; } diff --git a/src/compiler/php_generator.h b/src/compiler/php_generator.h index 905dc909a9..c3061f178e 100644 --- a/src/compiler/php_generator.h +++ b/src/compiler/php_generator.h @@ -38,7 +38,8 @@ namespace grpc_php_generator { -grpc::string GenerateFile(const grpc::protobuf::FileDescriptor *file); +grpc::string GenerateFile(const grpc::protobuf::FileDescriptor *file, + const grpc::protobuf::ServiceDescriptor *service); } // namespace grpc_php_generator diff --git a/src/compiler/php_generator_helpers.h b/src/compiler/php_generator_helpers.h index 61c4d21fff..97eb2d3e70 100644 --- a/src/compiler/php_generator_helpers.h +++ b/src/compiler/php_generator_helpers.h @@ -41,14 +41,23 @@ namespace grpc_php_generator { -inline grpc::string GetPHPServiceFilename(const grpc::string& filename) { - return grpc_generator::StripProto(filename) + "_grpc_pb.php"; +inline grpc::string GetPHPServiceFilename( + const grpc::protobuf::FileDescriptor *file, + const grpc::protobuf::ServiceDescriptor *service) { + std::vector<grpc::string> tokens = + grpc_generator::tokenize(file->package(), "."); + std::ostringstream oss; + for (unsigned int i = 0; i < tokens.size(); i++) { + oss << (i == 0 ? "" : "/") + << grpc_generator::CapitalizeFirstLetter(tokens[i]); + } + return oss.str() + "/" + service->name() + "Client.php"; } // Get leading or trailing comments in a string. Comment lines start with "// ". // Leading detached comments are put in in front of leading comments. template <typename DescriptorType> -inline grpc::string GetPHPComments(const DescriptorType* desc, +inline grpc::string GetPHPComments(const DescriptorType *desc, grpc::string prefix) { return grpc_generator::GetPrefixedComments(desc, true, prefix); } diff --git a/src/compiler/php_plugin.cc b/src/compiler/php_plugin.cc index 88acad6524..00d4cd5a85 100644 --- a/src/compiler/php_plugin.cc +++ b/src/compiler/php_plugin.cc @@ -51,18 +51,22 @@ class PHPGrpcGenerator : public grpc::protobuf::compiler::CodeGenerator { const grpc::string ¶meter, grpc::protobuf::compiler::GeneratorContext *context, grpc::string *error) const { - grpc::string code = GenerateFile(file); - if (code.size() == 0) { + if (file->service_count() == 0) { return true; } - // Get output file name - grpc::string file_name = GetPHPServiceFilename(file->name()); + for (int i = 0; i < file->service_count(); i++) { + grpc::string code = GenerateFile(file, file->service(i)); + + // Get output file name + grpc::string file_name = GetPHPServiceFilename(file, file->service(i)); + + std::unique_ptr<grpc::protobuf::io::ZeroCopyOutputStream> output( + context->Open(file_name)); + grpc::protobuf::io::CodedOutputStream coded_out(output.get()); + coded_out.WriteRaw(code.data(), code.size()); + } - std::unique_ptr<grpc::protobuf::io::ZeroCopyOutputStream> output( - context->Open(file_name)); - grpc::protobuf::io::CodedOutputStream coded_out(output.get()); - coded_out.WriteRaw(code.data(), code.size()); return true; } }; diff --git a/src/core/ext/client_channel/connector.c b/src/core/ext/client_channel/connector.c index 0582e5b372..7a720fd1bd 100644 --- a/src/core/ext/client_channel/connector.c +++ b/src/core/ext/client_channel/connector.c @@ -49,7 +49,7 @@ void grpc_connector_connect(grpc_exec_ctx* exec_ctx, grpc_connector* connector, connector->vtable->connect(exec_ctx, connector, in_args, out_args, notify); } -void grpc_connector_shutdown(grpc_exec_ctx* exec_ctx, - grpc_connector* connector) { - connector->vtable->shutdown(exec_ctx, connector); +void grpc_connector_shutdown(grpc_exec_ctx* exec_ctx, grpc_connector* connector, + grpc_error* why) { + connector->vtable->shutdown(exec_ctx, connector, why); } diff --git a/src/core/ext/client_channel/connector.h b/src/core/ext/client_channel/connector.h index 395f89b3b2..9bff41f003 100644 --- a/src/core/ext/client_channel/connector.h +++ b/src/core/ext/client_channel/connector.h @@ -68,7 +68,8 @@ struct grpc_connector_vtable { void (*ref)(grpc_connector *connector); void (*unref)(grpc_exec_ctx *exec_ctx, grpc_connector *connector); /** Implementation of grpc_connector_shutdown */ - void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_connector *connector); + void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_connector *connector, + grpc_error *why); /** Implementation of grpc_connector_connect */ void (*connect)(grpc_exec_ctx *exec_ctx, grpc_connector *connector, const grpc_connect_in_args *in_args, @@ -83,7 +84,7 @@ void grpc_connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *connector, grpc_connect_out_args *out_args, grpc_closure *notify); /** Cancel any pending connection */ -void grpc_connector_shutdown(grpc_exec_ctx *exec_ctx, - grpc_connector *connector); +void grpc_connector_shutdown(grpc_exec_ctx *exec_ctx, grpc_connector *connector, + grpc_error *why); #endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_CONNECTOR_H */ diff --git a/src/core/ext/client_channel/http_connect_handshaker.c b/src/core/ext/client_channel/http_connect_handshaker.c index 622d236320..58ab233f1b 100644 --- a/src/core/ext/client_channel/http_connect_handshaker.c +++ b/src/core/ext/client_channel/http_connect_handshaker.c @@ -123,7 +123,8 @@ static void handshake_failed_locked(grpc_exec_ctx* exec_ctx, // before destroying them, even if we know that there are no // pending read/write callbacks. This should be fixed, at which // point this can be removed. - grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint); + grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint, + GRPC_ERROR_REF(error)); // Not shutting down, so the handshake failed. Clean up before // invoking the callback. cleanup_args_for_failure_locked(exec_ctx, handshaker); @@ -251,15 +252,18 @@ static void http_connect_handshaker_destroy(grpc_exec_ctx* exec_ctx, } static void http_connect_handshaker_shutdown(grpc_exec_ctx* exec_ctx, - grpc_handshaker* handshaker_in) { + grpc_handshaker* handshaker_in, + grpc_error* why) { http_connect_handshaker* handshaker = (http_connect_handshaker*)handshaker_in; gpr_mu_lock(&handshaker->mu); if (!handshaker->shutdown) { handshaker->shutdown = true; - grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint); + grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint, + GRPC_ERROR_REF(why)); cleanup_args_for_failure_locked(exec_ctx, handshaker); } gpr_mu_unlock(&handshaker->mu); + GRPC_ERROR_UNREF(why); } static void http_connect_handshaker_do_handshake( diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c index c5041f6924..f1e4e079e2 100644 --- a/src/core/ext/client_channel/subchannel.c +++ b/src/core/ext/client_channel/subchannel.c @@ -273,7 +273,8 @@ static void disconnect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) { gpr_mu_lock(&c->mu); GPR_ASSERT(!c->disconnected); c->disconnected = true; - grpc_connector_shutdown(exec_ctx, c->connector); + grpc_connector_shutdown(exec_ctx, c->connector, + GRPC_ERROR_CREATE("Subchannel disconnected")); con = GET_CONNECTED_SUBCHANNEL(c, no_barrier); if (con != NULL) { GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, con, "connection"); diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.c b/src/core/ext/transport/chttp2/client/chttp2_connector.c index 013c96dc70..d0a762a280 100644 --- a/src/core/ext/transport/chttp2/client/chttp2_connector.c +++ b/src/core/ext/transport/chttp2/client/chttp2_connector.c @@ -92,19 +92,21 @@ static void chttp2_connector_unref(grpc_exec_ctx *exec_ctx, } static void chttp2_connector_shutdown(grpc_exec_ctx *exec_ctx, - grpc_connector *con) { + grpc_connector *con, grpc_error *why) { chttp2_connector *c = (chttp2_connector *)con; gpr_mu_lock(&c->mu); c->shutdown = true; if (c->handshake_mgr != NULL) { - grpc_handshake_manager_shutdown(exec_ctx, c->handshake_mgr); + grpc_handshake_manager_shutdown(exec_ctx, c->handshake_mgr, + GRPC_ERROR_REF(why)); } // If handshaking is not yet in progress, shutdown the endpoint. // Otherwise, the handshaker will do this for us. if (!c->connecting && c->endpoint != NULL) { - grpc_endpoint_shutdown(exec_ctx, c->endpoint); + grpc_endpoint_shutdown(exec_ctx, c->endpoint, GRPC_ERROR_REF(why)); } gpr_mu_unlock(&c->mu); + GRPC_ERROR_UNREF(why); } static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, @@ -121,7 +123,7 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, // before destroying them, even if we know that there are no // pending read/write callbacks. This should be fixed, at which // point this can be removed. - grpc_endpoint_shutdown(exec_ctx, args->endpoint); + grpc_endpoint_shutdown(exec_ctx, args->endpoint, GRPC_ERROR_REF(error)); grpc_endpoint_destroy(exec_ctx, args->endpoint); grpc_channel_args_destroy(exec_ctx, args->args); grpc_slice_buffer_destroy_internal(exec_ctx, args->read_buffer); @@ -195,7 +197,9 @@ static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_closure *notify = c->notify; c->notify = NULL; grpc_closure_sched(exec_ctx, notify, error); - if (c->endpoint != NULL) grpc_endpoint_shutdown(exec_ctx, c->endpoint); + if (c->endpoint != NULL) { + grpc_endpoint_shutdown(exec_ctx, c->endpoint, GRPC_ERROR_REF(error)); + } gpr_mu_unlock(&c->mu); chttp2_connector_unref(exec_ctx, arg); } else { diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.c b/src/core/ext/transport/chttp2/server/chttp2_server.c index 56a1a0de9b..ae2c3838ed 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.c +++ b/src/core/ext/transport/chttp2/server/chttp2_server.c @@ -101,16 +101,19 @@ static void pending_handshake_manager_remove_locked( } static void pending_handshake_manager_shutdown_locked(grpc_exec_ctx *exec_ctx, - server_state *state) { + server_state *state, + grpc_error *why) { pending_handshake_manager_node *prev_node = NULL; for (pending_handshake_manager_node *node = state->pending_handshake_mgrs; node != NULL; node = node->next) { - grpc_handshake_manager_shutdown(exec_ctx, node->handshake_mgr); + grpc_handshake_manager_shutdown(exec_ctx, node->handshake_mgr, + GRPC_ERROR_REF(why)); gpr_free(prev_node); prev_node = node; } gpr_free(prev_node); state->pending_handshake_mgrs = NULL; + GRPC_ERROR_UNREF(why); } static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, @@ -129,7 +132,7 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, // before destroying them, even if we know that there are no // pending read/write callbacks. This should be fixed, at which // point this can be removed. - grpc_endpoint_shutdown(exec_ctx, args->endpoint); + grpc_endpoint_shutdown(exec_ctx, args->endpoint, GRPC_ERROR_NONE); grpc_endpoint_destroy(exec_ctx, args->endpoint); grpc_channel_args_destroy(exec_ctx, args->args); grpc_slice_buffer_destroy_internal(exec_ctx, args->read_buffer); @@ -210,7 +213,8 @@ static void tcp_server_shutdown_complete(grpc_exec_ctx *exec_ctx, void *arg, gpr_mu_lock(&state->mu); grpc_closure *destroy_done = state->server_destroy_listener_done; GPR_ASSERT(state->shutdown); - pending_handshake_manager_shutdown_locked(exec_ctx, state); + pending_handshake_manager_shutdown_locked(exec_ctx, state, + GRPC_ERROR_REF(error)); gpr_mu_unlock(&state->mu); // Flush queued work before destroying handshaker factory, since that // may do a synchronous unref. diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 2004bc6437..15f486d676 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -417,7 +417,7 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, t->closed = 1; connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "close_transport"); - grpc_endpoint_shutdown(exec_ctx, t->ep); + grpc_endpoint_shutdown(exec_ctx, t->ep, GRPC_ERROR_REF(error)); /* flush writable stream list to avoid dangling references */ grpc_chttp2_stream *s; diff --git a/src/core/lib/channel/handshaker.c b/src/core/lib/channel/handshaker.c index c052ca5385..5bed2d041d 100644 --- a/src/core/lib/channel/handshaker.c +++ b/src/core/lib/channel/handshaker.c @@ -55,8 +55,8 @@ void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx, } void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx, - grpc_handshaker* handshaker) { - handshaker->vtable->shutdown(exec_ctx, handshaker); + grpc_handshaker* handshaker, grpc_error* why) { + handshaker->vtable->shutdown(exec_ctx, handshaker, why); } void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx, @@ -141,14 +141,17 @@ void grpc_handshake_manager_destroy(grpc_exec_ctx* exec_ctx, } void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx, - grpc_handshake_manager* mgr) { + grpc_handshake_manager* mgr, + grpc_error* why) { gpr_mu_lock(&mgr->mu); // Shutdown the handshaker that's currently in progress, if any. if (!mgr->shutdown && mgr->index > 0) { mgr->shutdown = true; - grpc_handshaker_shutdown(exec_ctx, mgr->handshakers[mgr->index - 1]); + grpc_handshaker_shutdown(exec_ctx, mgr->handshakers[mgr->index - 1], + GRPC_ERROR_REF(why)); } gpr_mu_unlock(&mgr->mu); + GRPC_ERROR_UNREF(why); } // Helper function to call either the next handshaker or the @@ -197,7 +200,8 @@ static void call_next_handshaker(grpc_exec_ctx* exec_ctx, void* arg, static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { grpc_handshake_manager* mgr = arg; if (error == GRPC_ERROR_NONE) { // Timer fired, rather than being cancelled. - grpc_handshake_manager_shutdown(exec_ctx, mgr); + grpc_handshake_manager_shutdown(exec_ctx, mgr, + GRPC_ERROR_CREATE("Handshake timed out")); } grpc_handshake_manager_unref(exec_ctx, mgr); } diff --git a/src/core/lib/channel/handshaker.h b/src/core/lib/channel/handshaker.h index 450b7adaee..a8e3692add 100644 --- a/src/core/lib/channel/handshaker.h +++ b/src/core/lib/channel/handshaker.h @@ -86,7 +86,8 @@ typedef struct { /// Shuts down the handshaker (e.g., to clean up when the operation is /// aborted in the middle). - void (*shutdown)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker); + void (*shutdown)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker, + grpc_error* why); /// Performs handshaking, modifying \a args as needed (e.g., to /// replace \a endpoint with a wrapped endpoint). @@ -111,7 +112,7 @@ void grpc_handshaker_init(const grpc_handshaker_vtable* vtable, void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker); void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx, - grpc_handshaker* handshaker); + grpc_handshaker* handshaker, grpc_error* why); void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker, grpc_tcp_server_acceptor* acceptor, @@ -141,7 +142,8 @@ void grpc_handshake_manager_destroy(grpc_exec_ctx* exec_ctx, /// The caller must still call grpc_handshake_manager_destroy() after /// calling this function. void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx, - grpc_handshake_manager* mgr); + grpc_handshake_manager* mgr, + grpc_error* why); /// Invokes handshakers in the order they were added. /// Takes ownership of \a endpoint, and then passes that ownership to diff --git a/src/core/lib/iomgr/endpoint.c b/src/core/lib/iomgr/endpoint.c index 2d300f4560..bf6e98146a 100644 --- a/src/core/lib/iomgr/endpoint.c +++ b/src/core/lib/iomgr/endpoint.c @@ -54,8 +54,9 @@ void grpc_endpoint_add_to_pollset_set(grpc_exec_ctx* exec_ctx, ep->vtable->add_to_pollset_set(exec_ctx, ep, pollset_set); } -void grpc_endpoint_shutdown(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep) { - ep->vtable->shutdown(exec_ctx, ep); +void grpc_endpoint_shutdown(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep, + grpc_error* why) { + ep->vtable->shutdown(exec_ctx, ep, why); } void grpc_endpoint_destroy(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep) { diff --git a/src/core/lib/iomgr/endpoint.h b/src/core/lib/iomgr/endpoint.h index 1609b64f2b..740357ecc5 100644 --- a/src/core/lib/iomgr/endpoint.h +++ b/src/core/lib/iomgr/endpoint.h @@ -57,7 +57,7 @@ struct grpc_endpoint_vtable { grpc_pollset *pollset); void (*add_to_pollset_set)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, grpc_pollset_set *pollset); - void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep); + void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, grpc_error *why); void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep); grpc_resource_user *(*get_resource_user)(grpc_endpoint *ep); char *(*get_peer)(grpc_endpoint *ep); @@ -96,7 +96,8 @@ void grpc_endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, /* Causes any pending and future read/write callbacks to run immediately with success==0 */ -void grpc_endpoint_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep); +void grpc_endpoint_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_error *why); void grpc_endpoint_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep); /* Add an endpoint to a pollset, so that when the pollset is polled, events from diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c index 39b4d9cbd7..51842fc208 100644 --- a/src/core/lib/iomgr/ev_epoll_linux.c +++ b/src/core/lib/iomgr/ev_epoll_linux.c @@ -143,6 +143,7 @@ struct grpc_fd { /* Indicates that the fd is shutdown and that any pending read/write closures should fail */ bool shutdown; + grpc_error *shutdown_error; /* reason for shutdown: set iff shutdown==true */ /* The fd is either closed or we relinquished control of it. In either cases, this indicates that the 'fd' on this structure is no longer valid */ @@ -907,6 +908,7 @@ static void unref_by(grpc_fd *fd, int n) { fd->freelist_next = fd_freelist; fd_freelist = fd; grpc_iomgr_unregister_object(&fd->iomgr_object); + if (fd->shutdown) GRPC_ERROR_UNREF(fd->shutdown_error); gpr_mu_unlock(&fd_freelist_mu); } else { @@ -1058,11 +1060,11 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, GRPC_ERROR_UNREF(error); } -static grpc_error *fd_shutdown_error(bool shutdown) { - if (!shutdown) { +static grpc_error *fd_shutdown_error(grpc_fd *fd) { + if (!fd->shutdown) { return GRPC_ERROR_NONE; } else { - return GRPC_ERROR_CREATE("FD shutdown"); + return GRPC_ERROR_CREATE_REFERENCING("FD shutdown", &fd->shutdown_error, 1); } } @@ -1076,7 +1078,7 @@ static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd, } else if (*st == CLOSURE_READY) { /* already ready ==> queue the closure to run immediately */ *st = CLOSURE_NOT_READY; - grpc_closure_sched(exec_ctx, closure, fd_shutdown_error(fd->shutdown)); + grpc_closure_sched(exec_ctx, closure, fd_shutdown_error(fd)); } else { /* upcallptr was set to a different closure. This is an error! */ gpr_log(GPR_ERROR, @@ -1098,7 +1100,7 @@ static int set_ready_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd, return 0; } else { /* waiting ==> queue closure */ - grpc_closure_sched(exec_ctx, *st, fd_shutdown_error(fd->shutdown)); + grpc_closure_sched(exec_ctx, *st, fd_shutdown_error(fd)); *st = CLOSURE_NOT_READY; return 1; } @@ -1123,17 +1125,20 @@ static bool fd_is_shutdown(grpc_fd *fd) { } /* Might be called multiple times */ -static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) { +static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) { gpr_mu_lock(&fd->po.mu); /* Do the actual shutdown only once */ if (!fd->shutdown) { fd->shutdown = true; + fd->shutdown_error = why; shutdown(fd->fd, SHUT_RDWR); /* Flush any pending read and write closures. Since fd->shutdown is 'true' at this point, the closures would be called with 'success = false' */ set_ready_locked(exec_ctx, fd, &fd->read_closure); set_ready_locked(exec_ctx, fd, &fd->write_closure); + } else { + GRPC_ERROR_UNREF(why); } gpr_mu_unlock(&fd->po.mu); } diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c index 9477ac3688..ca12932219 100644 --- a/src/core/lib/iomgr/ev_poll_posix.c +++ b/src/core/lib/iomgr/ev_poll_posix.c @@ -82,6 +82,7 @@ struct grpc_fd { int shutdown; int closed; int released; + grpc_error *shutdown_error; /* The watcher list. @@ -306,6 +307,7 @@ static void unref_by(grpc_fd *fd, int n) { if (old == n) { gpr_mu_destroy(&fd->mu); grpc_iomgr_unregister_object(&fd->iomgr_object); + if (fd->shutdown) GRPC_ERROR_UNREF(fd->shutdown_error); gpr_free(fd); } else { GPR_ASSERT(old > n); @@ -444,11 +446,11 @@ static void fd_ref(grpc_fd *fd) { ref_by(fd, 2); } static void fd_unref(grpc_fd *fd) { unref_by(fd, 2); } #endif -static grpc_error *fd_shutdown_error(bool shutdown) { - if (!shutdown) { +static grpc_error *fd_shutdown_error(grpc_fd *fd) { + if (!fd->shutdown) { return GRPC_ERROR_NONE; } else { - return GRPC_ERROR_CREATE("FD shutdown"); + return GRPC_ERROR_CREATE_REFERENCING("FD shutdown", &fd->shutdown_error, 1); } } @@ -462,7 +464,7 @@ static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd, } else if (*st == CLOSURE_READY) { /* already ready ==> queue the closure to run immediately */ *st = CLOSURE_NOT_READY; - grpc_closure_sched(exec_ctx, closure, fd_shutdown_error(fd->shutdown)); + grpc_closure_sched(exec_ctx, closure, fd_shutdown_error(fd)); maybe_wake_one_watcher_locked(fd); } else { /* upcallptr was set to a different closure. This is an error! */ @@ -485,7 +487,7 @@ static int set_ready_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd, return 0; } else { /* waiting ==> queue closure */ - grpc_closure_sched(exec_ctx, *st, fd_shutdown_error(fd->shutdown)); + grpc_closure_sched(exec_ctx, *st, fd_shutdown_error(fd)); *st = CLOSURE_NOT_READY; return 1; } @@ -496,15 +498,18 @@ static void set_read_notifier_pollset_locked( fd->read_notifier_pollset = read_notifier_pollset; } -static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) { +static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) { gpr_mu_lock(&fd->mu); /* only shutdown once */ if (!fd->shutdown) { fd->shutdown = 1; + fd->shutdown_error = why; /* signal read/write closed to OS so that future operations fail */ shutdown(fd->fd, SHUT_RDWR); set_ready_locked(exec_ctx, fd, &fd->read_closure); set_ready_locked(exec_ctx, fd, &fd->write_closure); + } else { + GRPC_ERROR_UNREF(why); } gpr_mu_unlock(&fd->mu); } diff --git a/src/core/lib/iomgr/ev_posix.c b/src/core/lib/iomgr/ev_posix.c index c106ba5400..5bb55631d6 100644 --- a/src/core/lib/iomgr/ev_posix.c +++ b/src/core/lib/iomgr/ev_posix.c @@ -162,8 +162,8 @@ void grpc_fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure *on_done, g_event_engine->fd_orphan(exec_ctx, fd, on_done, release_fd, reason); } -void grpc_fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) { - g_event_engine->fd_shutdown(exec_ctx, fd); +void grpc_fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) { + g_event_engine->fd_shutdown(exec_ctx, fd, why); } bool grpc_fd_is_shutdown(grpc_fd *fd) { diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h index 1068a4bad5..a589efdeec 100644 --- a/src/core/lib/iomgr/ev_posix.h +++ b/src/core/lib/iomgr/ev_posix.h @@ -51,7 +51,7 @@ typedef struct grpc_event_engine_vtable { int (*fd_wrapped_fd)(grpc_fd *fd); void (*fd_orphan)(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure *on_done, int *release_fd, const char *reason); - void (*fd_shutdown)(grpc_exec_ctx *exec_ctx, grpc_fd *fd); + void (*fd_shutdown)(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why); void (*fd_notify_on_read)(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure *closure); void (*fd_notify_on_write)(grpc_exec_ctx *exec_ctx, grpc_fd *fd, @@ -140,7 +140,7 @@ void grpc_fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure *on_done, bool grpc_fd_is_shutdown(grpc_fd *fd); /* Cause any current and future callbacks to fail. */ -void grpc_fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd); +void grpc_fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why); /* Register read interest, causing read_cb to be called once when fd becomes readable, on deadline specified by deadline, or on shutdown triggered by diff --git a/src/core/lib/iomgr/network_status_tracker.c b/src/core/lib/iomgr/network_status_tracker.c index a5ca9ed2c3..1601a39002 100644 --- a/src/core/lib/iomgr/network_status_tracker.c +++ b/src/core/lib/iomgr/network_status_tracker.c @@ -117,7 +117,8 @@ void grpc_network_status_shutdown_all_endpoints() { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; for (endpoint_ll_node *curr = head; curr != NULL; curr = curr->next) { - curr->ep->vtable->shutdown(&exec_ctx, curr->ep); + curr->ep->vtable->shutdown(&exec_ctx, curr->ep, + GRPC_ERROR_CREATE("Network unavailable")); } gpr_mu_unlock(&g_endpoint_mutex); grpc_exec_ctx_finish(&exec_ctx); diff --git a/src/core/lib/iomgr/tcp_client_posix.c b/src/core/lib/iomgr/tcp_client_posix.c index 16b0f4e73c..0144192b71 100644 --- a/src/core/lib/iomgr/tcp_client_posix.c +++ b/src/core/lib/iomgr/tcp_client_posix.c @@ -121,7 +121,8 @@ static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { } gpr_mu_lock(&ac->mu); if (ac->fd != NULL) { - grpc_fd_shutdown(exec_ctx, ac->fd); + grpc_fd_shutdown(exec_ctx, ac->fd, + GRPC_ERROR_CREATE("connect() timed out")); } done = (--ac->refs == 0); gpr_mu_unlock(&ac->mu); diff --git a/src/core/lib/iomgr/tcp_posix.c b/src/core/lib/iomgr/tcp_posix.c index a33e63e845..a4381f8fc9 100644 --- a/src/core/lib/iomgr/tcp_posix.c +++ b/src/core/lib/iomgr/tcp_posix.c @@ -119,9 +119,10 @@ static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */, static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */, grpc_error *error); -static void tcp_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) { +static void tcp_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_error *why) { grpc_tcp *tcp = (grpc_tcp *)ep; - grpc_fd_shutdown(exec_ctx, tcp->em_fd); + grpc_fd_shutdown(exec_ctx, tcp->em_fd, why); grpc_resource_user_shutdown(exec_ctx, tcp->resource_user); } diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c index 20efb678b2..e9e7511c9c 100644 --- a/src/core/lib/iomgr/tcp_server_posix.c +++ b/src/core/lib/iomgr/tcp_server_posix.c @@ -276,7 +276,8 @@ static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { if (s->active_ports) { grpc_tcp_listener *sp; for (sp = s->head; sp; sp = sp->next) { - grpc_fd_shutdown(exec_ctx, sp->emfd); + grpc_fd_shutdown(exec_ctx, sp->emfd, + GRPC_ERROR_CREATE("Server destroyed")); } gpr_mu_unlock(&s->mu); } else { @@ -773,7 +774,8 @@ void grpc_tcp_server_shutdown_listeners(grpc_exec_ctx *exec_ctx, if (s->active_ports) { grpc_tcp_listener *sp; for (sp = s->head; sp; sp = sp->next) { - grpc_fd_shutdown(exec_ctx, sp->emfd); + grpc_fd_shutdown(exec_ctx, sp->emfd, + GRPC_ERROR_CREATE("Server shutdown")); } } gpr_mu_unlock(&s->mu); diff --git a/src/core/lib/iomgr/tcp_uv.c b/src/core/lib/iomgr/tcp_uv.c index 7f4ea49a1c..5fb398c50b 100644 --- a/src/core/lib/iomgr/tcp_uv.c +++ b/src/core/lib/iomgr/tcp_uv.c @@ -298,13 +298,15 @@ static void uv_add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, static void shutdown_callback(uv_shutdown_t *req, int status) {} -static void uv_endpoint_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) { +static void uv_endpoint_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_error *why) { grpc_tcp *tcp = (grpc_tcp *)ep; if (!tcp->shutting_down) { tcp->shutting_down = true; uv_shutdown_t *req = &tcp->shutdown_req; uv_shutdown(req, (uv_stream_t *)tcp->handle, shutdown_callback); } + GRPC_ERROR_UNREF(why); } static void uv_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) { diff --git a/src/core/lib/iomgr/tcp_windows.c b/src/core/lib/iomgr/tcp_windows.c index 84f791ba07..6c413971e3 100644 --- a/src/core/lib/iomgr/tcp_windows.c +++ b/src/core/lib/iomgr/tcp_windows.c @@ -116,6 +116,7 @@ typedef struct grpc_tcp { to protect ourselves when requesting a shutdown. */ gpr_mu mu; int shutting_down; + grpc_error *shutdown_error; char *peer_string; } grpc_tcp; @@ -125,6 +126,7 @@ static void tcp_free(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) { gpr_mu_destroy(&tcp->mu); gpr_free(tcp->peer_string); grpc_resource_user_unref(exec_ctx, tcp->resource_user); + if (tcp->shutting_down) GRPC_ERROR_UNREF(tcp->shutdown_error); gpr_free(tcp); } @@ -182,7 +184,10 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) { grpc_slice_buffer_add(tcp->read_slices, sub); } else { grpc_slice_unref_internal(exec_ctx, tcp->read_slice); - error = GRPC_ERROR_CREATE("End of TCP stream"); + error = tcp->shutting_down + ? GRPC_ERROR_CREATE_REFERENCING("TCP stream shutting down", + &tcp->shutdown_error, 1) + : GRPC_ERROR_CREATE("End of TCP stream"); } } } @@ -203,8 +208,9 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, WSABUF buffer; if (tcp->shutting_down) { - grpc_closure_sched(exec_ctx, cb, - GRPC_ERROR_CREATE("TCP socket is shutting down")); + grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_CREATE_REFERENCING( + "TCP socket is shutting down", + &tcp->shutdown_error, 1)); return; } @@ -291,8 +297,9 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, size_t len; if (tcp->shutting_down) { - grpc_closure_sched(exec_ctx, cb, - GRPC_ERROR_CREATE("TCP socket is shutting down")); + grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_CREATE_REFERENCING( + "TCP socket is shutting down", + &tcp->shutdown_error, 1)); return; } @@ -373,12 +380,18 @@ static void win_add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, we're not going to protect against these. However the IO Completion Port callback will happen from another thread, so we need to protect against concurrent access of the data structure in that regard. */ -static void win_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) { +static void win_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_error *why) { grpc_tcp *tcp = (grpc_tcp *)ep; gpr_mu_lock(&tcp->mu); /* At that point, what may happen is that we're already inside the IOCP callback. See the comments in on_read and on_write. */ - tcp->shutting_down = 1; + if (!tcp->shutting_down) { + tcp->shutting_down = 1; + tcp->shutdown_error = why; + } else { + GRPC_ERROR_UNREF(why); + } grpc_winsocket_shutdown(tcp->socket); gpr_mu_unlock(&tcp->mu); grpc_resource_user_shutdown(exec_ctx, tcp->resource_user); diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index dfbd295c91..3b23b47d4f 100644 --- a/src/core/lib/iomgr/udp_server.c +++ b/src/core/lib/iomgr/udp_server.c @@ -203,7 +203,8 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, for (sp = s->head; sp; sp = sp->next) { GPR_ASSERT(sp->orphan_cb); sp->orphan_cb(sp->emfd); - grpc_fd_shutdown(exec_ctx, sp->emfd); + grpc_fd_shutdown(exec_ctx, sp->emfd, + GRPC_ERROR_CREATE("Server destroyed")); } gpr_mu_unlock(&s->mu); } else { diff --git a/src/core/lib/security/transport/secure_endpoint.c b/src/core/lib/security/transport/secure_endpoint.c index 18a7a6f7e7..7d58843d69 100644 --- a/src/core/lib/security/transport/secure_endpoint.c +++ b/src/core/lib/security/transport/secure_endpoint.c @@ -341,10 +341,10 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, GPR_TIMER_END("secure_endpoint.endpoint_write", 0); } -static void endpoint_shutdown(grpc_exec_ctx *exec_ctx, - grpc_endpoint *secure_ep) { +static void endpoint_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, + grpc_error *why) { secure_endpoint *ep = (secure_endpoint *)secure_ep; - grpc_endpoint_shutdown(exec_ctx, ep->wrapped_ep); + grpc_endpoint_shutdown(exec_ctx, ep->wrapped_ep, why); } static void endpoint_destroy(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/security/transport/security_handshaker.c b/src/core/lib/security/transport/security_handshaker.c index 37d57d759b..bb8a3bf6cd 100644 --- a/src/core/lib/security/transport/security_handshaker.c +++ b/src/core/lib/security/transport/security_handshaker.c @@ -130,7 +130,7 @@ static void security_handshake_failed_locked(grpc_exec_ctx *exec_ctx, // before destroying them, even if we know that there are no // pending read/write callbacks. This should be fixed, at which // point this can be removed. - grpc_endpoint_shutdown(exec_ctx, h->args->endpoint); + grpc_endpoint_shutdown(exec_ctx, h->args->endpoint, GRPC_ERROR_REF(error)); // Not shutting down, so the write failed. Clean up before // invoking the callback. cleanup_args_for_failure_locked(exec_ctx, h); @@ -347,15 +347,17 @@ static void security_handshaker_destroy(grpc_exec_ctx *exec_ctx, } static void security_handshaker_shutdown(grpc_exec_ctx *exec_ctx, - grpc_handshaker *handshaker) { + grpc_handshaker *handshaker, + grpc_error *why) { security_handshaker *h = (security_handshaker *)handshaker; gpr_mu_lock(&h->mu); if (!h->shutdown) { h->shutdown = true; - grpc_endpoint_shutdown(exec_ctx, h->args->endpoint); + grpc_endpoint_shutdown(exec_ctx, h->args->endpoint, GRPC_ERROR_REF(why)); cleanup_args_for_failure_locked(exec_ctx, h); } gpr_mu_unlock(&h->mu); + GRPC_ERROR_UNREF(why); } static void security_handshaker_do_handshake(grpc_exec_ctx *exec_ctx, @@ -417,7 +419,10 @@ static void fail_handshaker_destroy(grpc_exec_ctx *exec_ctx, } static void fail_handshaker_shutdown(grpc_exec_ctx *exec_ctx, - grpc_handshaker *handshaker) {} + grpc_handshaker *handshaker, + grpc_error *why) { + GRPC_ERROR_UNREF(why); +} static void fail_handshaker_do_handshake(grpc_exec_ctx *exec_ctx, grpc_handshaker *handshaker, diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index dcdddc769e..6bf9756962 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -336,6 +336,22 @@ namespace Grpc.Core.Tests } [Test] + public void StatusDetailIsUtf8() + { + // some japanese and chinese characters + var nonAsciiString = "\u30a1\u30a2\u30a3 \u62b5\u6297\u662f\u5f92\u52b3\u7684"; + helper.UnaryHandler = new UnaryServerMethod<string, string>(async (request, context) => + { + context.Status = new Status(StatusCode.Unknown, nonAsciiString); + return ""; + }); + + var ex = Assert.Throws<RpcException>(() => Calls.BlockingUnaryCall(helper.CreateUnaryCall(), "abc")); + Assert.AreEqual(StatusCode.Unknown, ex.Status.StatusCode); + Assert.AreEqual(nonAsciiString, ex.Status.Detail); + } + + [Test] public void ServerCallContext_PeerInfoPresent() { helper.UnaryHandler = new UnaryServerMethod<string, string>(async (request, context) => diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs index 0e4a77be81..efae149f09 100644 --- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs @@ -33,6 +33,7 @@ using System; using System.Runtime.InteropServices; +using System.Text; using Grpc.Core; namespace Grpc.Core.Internal @@ -42,6 +43,7 @@ namespace Grpc.Core.Internal /// </summary> internal class BatchContextSafeHandle : SafeHandleZeroIsInvalid { + static readonly Encoding EncodingUTF8 = System.Text.Encoding.UTF8; static readonly NativeMethods Native = NativeMethods.Get(); private BatchContextSafeHandle() @@ -73,7 +75,7 @@ namespace Grpc.Core.Internal { UIntPtr detailsLength; IntPtr detailsPtr = Native.grpcsharp_batch_context_recv_status_on_client_details(this, out detailsLength); - string details = Marshal.PtrToStringAnsi(detailsPtr, (int) detailsLength.ToUInt32()); + string details = PtrToStringUtf8(detailsPtr, (int) detailsLength.ToUInt32()); var status = new Status(Native.grpcsharp_batch_context_recv_status_on_client_status(this), details); IntPtr metadataArrayPtr = Native.grpcsharp_batch_context_recv_status_on_client_trailing_metadata(this); @@ -106,5 +108,12 @@ namespace Grpc.Core.Internal Native.grpcsharp_batch_context_destroy(handle); return true; } + + string PtrToStringUtf8(IntPtr ptr, int len) + { + var bytes = new byte[len]; + Marshal.Copy(ptr, bytes, 0, len); + return EncodingUTF8.GetString(bytes); + } } } diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs index 6bfcc7fa74..710ca480e8 100644 --- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs @@ -32,6 +32,7 @@ using System; using System.Diagnostics; using System.Runtime.InteropServices; +using System.Text; using Grpc.Core; using Grpc.Core.Utils; using Grpc.Core.Profiling; @@ -44,6 +45,7 @@ namespace Grpc.Core.Internal internal class CallSafeHandle : SafeHandleZeroIsInvalid, INativeCall { public static readonly CallSafeHandle NullInstance = new CallSafeHandle(); + static readonly Encoding EncodingUTF8 = System.Text.Encoding.UTF8; static readonly NativeMethods Native = NativeMethods.Get(); const uint GRPC_WRITE_BUFFER_HINT = 1; @@ -138,7 +140,8 @@ namespace Grpc.Core.Internal var ctx = BatchContextSafeHandle.Create(); var optionalPayloadLength = optionalPayload != null ? new UIntPtr((ulong)optionalPayload.Length) : UIntPtr.Zero; completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success)); - Native.grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail, metadataArray, sendEmptyInitialMetadata, + var statusDetailBytes = EncodingUTF8.GetBytes(status.Detail); + Native.grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, statusDetailBytes, new UIntPtr((ulong)statusDetailBytes.Length), metadataArray, sendEmptyInitialMetadata, optionalPayload, optionalPayloadLength, writeFlags).CheckOk(); } } diff --git a/src/csharp/Grpc.Core/Internal/NativeMethods.cs b/src/csharp/Grpc.Core/Internal/NativeMethods.cs index 2f377071f7..aff9550e8d 100644 --- a/src/csharp/Grpc.Core/Internal/NativeMethods.cs +++ b/src/csharp/Grpc.Core/Internal/NativeMethods.cs @@ -336,7 +336,7 @@ namespace Grpc.Core.Internal public delegate CallError grpcsharp_call_send_close_from_client_delegate(CallSafeHandle call, BatchContextSafeHandle ctx); public delegate CallError grpcsharp_call_send_status_from_server_delegate(CallSafeHandle call, - BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, + BatchContextSafeHandle ctx, StatusCode statusCode, byte[] statusMessage, UIntPtr statusMessageLen, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, byte[] optionalSendBuffer, UIntPtr optionalSendBufferLen, WriteFlags writeFlags); public delegate CallError grpcsharp_call_recv_message_delegate(CallSafeHandle call, BatchContextSafeHandle ctx); diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c index e308b0229c..e1f4d7cdf3 100644 --- a/src/csharp/ext/grpc_csharp_ext.c +++ b/src/csharp/ext/grpc_csharp_ext.c @@ -734,14 +734,15 @@ grpcsharp_call_send_close_from_client(grpc_call *call, GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_send_status_from_server( grpc_call *call, grpcsharp_batch_context *ctx, grpc_status_code status_code, - const char *status_details, grpc_metadata_array *trailing_metadata, + const char *status_details, size_t status_details_len, + grpc_metadata_array *trailing_metadata, int32_t send_empty_initial_metadata, const char* optional_send_buffer, size_t optional_send_buffer_len, uint32_t write_flags) { /* TODO: don't use magic number */ grpc_op ops[3]; memset(ops, 0, sizeof(ops)); size_t nops = 1; - grpc_slice status_details_slice = grpc_slice_from_copied_string(status_details); + grpc_slice status_details_slice = grpc_slice_from_copied_buffer(status_details, status_details_len); ops[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER; ops[0].data.send_status_from_server.status = status_code; ops[0].data.send_status_from_server.status_details = &status_details_slice; diff --git a/src/php/composer.json b/src/php/composer.json index 2d5d555bc2..992f6ac3f6 100644 --- a/src/php/composer.json +++ b/src/php/composer.json @@ -1,14 +1,10 @@ { - "name": "grpc/grpc", - "type": "library", - "description": "gRPC library for PHP", - "keywords": ["rpc"], - "homepage": "http://grpc.io", + "name": "grpc/grpc-dev", + "description": "gRPC library for PHP - for Developement use only", "license": "BSD-3-Clause", "version": "1.1.0", "require": { "php": ">=5.5.0", - "ext-grpc": "*", "google/protobuf": "v3.1.0-alpha-1" }, "require-dev": { @@ -16,7 +12,11 @@ }, "autoload": { "psr-4": { - "Grpc\\": "lib/Grpc/" + "Grpc\\": "lib/Grpc/", + "Grpc\\Testing\\": "tests/interop/Grpc/Testing/", + "GPBMetadata\\Src\\Proto\\Grpc\\Testing\\": "tests/interop/GPBMetadata/Src/Proto/Grpc/Testing/", + "Math\\": "tests/generated_code/Math/", + "GPBMetadata\\": "tests/generated_code/GPBMetadata/" } } } diff --git a/src/php/tests/generated_code/AbstractGeneratedCodeTest.php b/src/php/tests/generated_code/AbstractGeneratedCodeTest.php index 8fe9bc26d8..c50b1c6943 100644 --- a/src/php/tests/generated_code/AbstractGeneratedCodeTest.php +++ b/src/php/tests/generated_code/AbstractGeneratedCodeTest.php @@ -32,8 +32,11 @@ * */ require_once realpath(dirname(__FILE__).'/../../vendor/autoload.php'); -require_once dirname(__FILE__).'/math.pb.php'; -require_once dirname(__FILE__).'/math_grpc_pb.php'; + +// The following includes are needed when using protobuf 3.1.0 +// and will suppress warnings when using protobuf 3.2.0+ +@include_once dirname(__FILE__).'/math.pb.php'; +@include_once dirname(__FILE__).'/math_grpc_pb.php'; abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase { @@ -70,7 +73,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testClose() { self::$client->close(); - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $call = self::$client->Div($div_arg); } @@ -79,20 +82,20 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase */ public function testInvalidMetadata() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $call = self::$client->Div($div_arg, [' ' => 'abc123']); } public function testGetCallMetadata() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $call = self::$client->Div($div_arg); $this->assertTrue(is_array($call->getMetadata())); } public function testTimeout() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $call = self::$client->Div($div_arg, [], ['timeout' => 1]); list($response, $status) = $call->wait(); $this->assertSame(\Grpc\STATUS_DEADLINE_EXCEEDED, $status->code); @@ -100,7 +103,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testCancel() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $call = self::$client->Div($div_arg); $call->cancel(); list($response, $status) = $call->wait(); @@ -109,7 +112,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testCallCredentialsCallback() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $call = self::$client->Div($div_arg, array(), array( 'call_credentials_callback' => function ($context) { return array(); @@ -122,7 +125,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testCallCredentialsCallback2() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $call = self::$client->Div($div_arg); $call_credentials = Grpc\CallCredentials::createFromPlugin( function ($context) { @@ -143,7 +146,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase $invalid_client = new DummyInvalidClient('host', [ 'credentials' => Grpc\ChannelCredentials::createInsecure(), ]); - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $invalid_client->InvalidUnaryCall($div_arg); } @@ -166,7 +169,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testWriteFlags() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $div_arg->setDividend(7); $div_arg->setDivisor(4); $call = self::$client->Div($div_arg, [], @@ -180,7 +183,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testWriteFlagsServerStreaming() { - $fib_arg = new math\FibArgs(); + $fib_arg = new Math\FibArgs(); $fib_arg->setLimit(7); $call = self::$client->Fib($fib_arg, [], ['flags' => Grpc\WRITE_NO_COMPRESS]); @@ -192,7 +195,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testWriteFlagsClientStreaming() { $call = self::$client->Sum(); - $num = new math\Num(); + $num = new Math\Num(); $num->setNum(1); $call->write($num, ['flags' => Grpc\WRITE_NO_COMPRESS]); list($response, $status) = $call->wait(); @@ -202,7 +205,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testWriteFlagsBidiStreaming() { $call = self::$client->DivMany(); - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $div_arg->setDividend(7); $div_arg->setDivisor(4); $call->write($div_arg, ['flags' => Grpc\WRITE_NO_COMPRESS]); @@ -214,7 +217,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testSimpleRequest() { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $div_arg->setDividend(7); $div_arg->setDivisor(4); $call = self::$client->Div($div_arg); @@ -227,7 +230,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase public function testServerStreaming() { - $fib_arg = new math\FibArgs(); + $fib_arg = new Math\FibArgs(); $fib_arg->setLimit(7); $call = self::$client->Fib($fib_arg); $this->assertTrue(is_string($call->getPeer())); @@ -246,7 +249,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase $call = self::$client->Sum(); $this->assertTrue(is_string($call->getPeer())); for ($i = 0; $i < 7; ++$i) { - $num = new math\Num(); + $num = new Math\Num(); $num->setNum($i); $call->write($num); } @@ -260,7 +263,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase $call = self::$client->DivMany(); $this->assertTrue(is_string($call->getPeer())); for ($i = 0; $i < 7; ++$i) { - $div_arg = new math\DivArgs(); + $div_arg = new Math\DivArgs(); $div_arg->setDividend(2 * $i + 1); $div_arg->setDivisor(2); $call->write($div_arg); @@ -276,7 +279,7 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase class DummyInvalidClient extends \Grpc\BaseStub { - public function InvalidUnaryCall(\math\DivArgs $argument, + public function InvalidUnaryCall(\Math\DivArgs $argument, $metadata = [], $options = []) { diff --git a/src/php/tests/generated_code/GeneratedCodeTest.php b/src/php/tests/generated_code/GeneratedCodeTest.php index 0cdce6cf92..12ba012910 100755 --- a/src/php/tests/generated_code/GeneratedCodeTest.php +++ b/src/php/tests/generated_code/GeneratedCodeTest.php @@ -37,7 +37,7 @@ class GeneratedCodeTest extends AbstractGeneratedCodeTest { public function setUp() { - self::$client = new math\MathClient( + self::$client = new Math\MathClient( getenv('GRPC_TEST_HOST'), [ 'credentials' => Grpc\ChannelCredentials::createInsecure(), ]); diff --git a/src/php/tests/generated_code/GeneratedCodeWithCallbackTest.php b/src/php/tests/generated_code/GeneratedCodeWithCallbackTest.php index 6b70b8ac10..e1899484ec 100644 --- a/src/php/tests/generated_code/GeneratedCodeWithCallbackTest.php +++ b/src/php/tests/generated_code/GeneratedCodeWithCallbackTest.php @@ -37,7 +37,7 @@ class GeneratedCodeWithCallbackTest extends AbstractGeneratedCodeTest { public function setUp() { - self::$client = new math\MathClient( + self::$client = new Math\MathClient( getenv('GRPC_TEST_HOST'), ['credentials' => Grpc\ChannelCredentials::createInsecure(), 'update_metadata' => function ($a_hash, diff --git a/src/php/tests/interop/interop_client.php b/src/php/tests/interop/interop_client.php index 2acf5612c7..cf93ac39e0 100755 --- a/src/php/tests/interop/interop_client.php +++ b/src/php/tests/interop/interop_client.php @@ -32,8 +32,12 @@ * */ require_once realpath(dirname(__FILE__).'/../../vendor/autoload.php'); -require 'src/proto/grpc/testing/test.pb.php'; -require 'src/proto/grpc/testing/test_grpc_pb.php'; + +// The following includes are needed when using protobuf 3.1.0 +// and will suppress warnings when using protobuf 3.2.0+ +@include_once 'src/proto/grpc/testing/test.pb.php'; +@include_once 'src/proto/grpc/testing/test_grpc_pb.php'; + use Google\Auth\CredentialsLoader; use Google\Auth\ApplicationDefaultCredentials; use GuzzleHttp\ClientInterface; @@ -70,7 +74,7 @@ function hardAssertIfStatusOk($status) function emptyUnary($stub) { list($result, $status) = - $stub->EmptyCall(new grpc\testing\EmptyMessage())->wait(); + $stub->EmptyCall(new Grpc\Testing\EmptyMessage())->wait(); hardAssertIfStatusOk($status); hardAssert($result !== null, 'Call completed with a null response'); } @@ -98,11 +102,11 @@ function performLargeUnary($stub, $fillUsername = false, $request_len = 271828; $response_len = 314159; - $request = new grpc\testing\SimpleRequest(); - $request->setResponseType(grpc\testing\PayloadType::COMPRESSABLE); + $request = new Grpc\Testing\SimpleRequest(); + $request->setResponseType(Grpc\Testing\PayloadType::COMPRESSABLE); $request->setResponseSize($response_len); - $payload = new grpc\testing\Payload(); - $payload->setType(grpc\testing\PayloadType::COMPRESSABLE); + $payload = new Grpc\Testing\Payload(); + $payload->setType(Grpc\Testing\PayloadType::COMPRESSABLE); $payload->setBody(str_repeat("\0", $request_len)); $request->setPayload($payload); $request->setFillUsername($fillUsername); @@ -117,7 +121,7 @@ function performLargeUnary($stub, $fillUsername = false, hardAssertIfStatusOk($status); hardAssert($result !== null, 'Call returned a null response'); $payload = $result->getPayload(); - hardAssert($payload->getType() === grpc\testing\PayloadType::COMPRESSABLE, + hardAssert($payload->getType() === Grpc\Testing\PayloadType::COMPRESSABLE, 'Payload had the wrong type'); hardAssert(strlen($payload->getBody()) === $response_len, 'Payload had the wrong length'); @@ -249,8 +253,8 @@ function clientStreaming($stub) $requests = array_map( function ($length) { - $request = new grpc\testing\StreamingInputCallRequest(); - $payload = new grpc\testing\Payload(); + $request = new Grpc\Testing\StreamingInputCallRequest(); + $payload = new Grpc\Testing\Payload(); $payload->setBody(str_repeat("\0", $length)); $request->setPayload($payload); @@ -276,10 +280,10 @@ function serverStreaming($stub) { $sizes = [31415, 9, 2653, 58979]; - $request = new grpc\testing\StreamingOutputCallRequest(); - $request->setResponseType(grpc\testing\PayloadType::COMPRESSABLE); + $request = new Grpc\Testing\StreamingOutputCallRequest(); + $request->setResponseType(Grpc\Testing\PayloadType::COMPRESSABLE); foreach ($sizes as $size) { - $response_parameters = new grpc\testing\ResponseParameters(); + $response_parameters = new Grpc\Testing\ResponseParameters(); $response_parameters->setSize($size); $request->getResponseParameters()[] = $response_parameters; } @@ -290,7 +294,7 @@ function serverStreaming($stub) hardAssert($i < 4, 'Too many responses'); $payload = $value->getPayload(); hardAssert( - $payload->getType() === grpc\testing\PayloadType::COMPRESSABLE, + $payload->getType() === Grpc\Testing\PayloadType::COMPRESSABLE, 'Payload '.$i.' had the wrong type'); hardAssert(strlen($payload->getBody()) === $sizes[$i], 'Response '.$i.' had the wrong length'); @@ -311,12 +315,12 @@ function pingPong($stub) $call = $stub->FullDuplexCall(); for ($i = 0; $i < 4; ++$i) { - $request = new grpc\testing\StreamingOutputCallRequest(); - $request->setResponseType(grpc\testing\PayloadType::COMPRESSABLE); - $response_parameters = new grpc\testing\ResponseParameters(); + $request = new Grpc\Testing\StreamingOutputCallRequest(); + $request->setResponseType(Grpc\Testing\PayloadType::COMPRESSABLE); + $response_parameters = new Grpc\Testing\ResponseParameters(); $response_parameters->setSize($response_lengths[$i]); $request->getResponseParameters()[] = $response_parameters; - $payload = new grpc\testing\Payload(); + $payload = new Grpc\Testing\Payload(); $payload->setBody(str_repeat("\0", $request_lengths[$i])); $request->setPayload($payload); @@ -326,7 +330,7 @@ function pingPong($stub) hardAssert($response !== null, 'Server returned too few responses'); $payload = $response->getPayload(); hardAssert( - $payload->getType() === grpc\testing\PayloadType::COMPRESSABLE, + $payload->getType() === Grpc\Testing\PayloadType::COMPRESSABLE, 'Payload '.$i.' had the wrong type'); hardAssert(strlen($payload->getBody()) === $response_lengths[$i], 'Payload '.$i.' had the wrong length'); @@ -371,12 +375,12 @@ function cancelAfterBegin($stub) function cancelAfterFirstResponse($stub) { $call = $stub->FullDuplexCall(); - $request = new grpc\testing\StreamingOutputCallRequest(); - $request->setResponseType(grpc\testing\PayloadType::COMPRESSABLE); - $response_parameters = new grpc\testing\ResponseParameters(); + $request = new Grpc\Testing\StreamingOutputCallRequest(); + $request->setResponseType(Grpc\Testing\PayloadType::COMPRESSABLE); + $response_parameters = new Grpc\Testing\ResponseParameters(); $response_parameters->setSize(31415); $request->getResponseParameters()[] = $response_parameters; - $payload = new grpc\testing\Payload(); + $payload = new Grpc\Testing\Payload(); $payload->setBody(str_repeat("\0", 27182)); $request->setPayload($payload); @@ -391,12 +395,12 @@ function cancelAfterFirstResponse($stub) function timeoutOnSleepingServer($stub) { $call = $stub->FullDuplexCall([], ['timeout' => 1000]); - $request = new grpc\testing\StreamingOutputCallRequest(); - $request->setResponseType(grpc\testing\PayloadType::COMPRESSABLE); - $response_parameters = new grpc\testing\ResponseParameters(); + $request = new Grpc\Testing\StreamingOutputCallRequest(); + $request->setResponseType(Grpc\Testing\PayloadType::COMPRESSABLE); + $response_parameters = new Grpc\Testing\ResponseParameters(); $response_parameters->setSize(8); $request->getResponseParameters()[] = $response_parameters; - $payload = new grpc\testing\Payload(); + $payload = new Grpc\Testing\Payload(); $payload->setBody(str_repeat("\0", 9)); $request->setPayload($payload); @@ -416,11 +420,11 @@ function customMetadata($stub) $request_len = 271828; $response_len = 314159; - $request = new grpc\testing\SimpleRequest(); - $request->setResponseType(grpc\testing\PayloadType::COMPRESSABLE); + $request = new Grpc\Testing\SimpleRequest(); + $request->setResponseType(Grpc\Testing\PayloadType::COMPRESSABLE); $request->setResponseSize($response_len); - $payload = new grpc\testing\Payload(); - $payload->setType(grpc\testing\PayloadType::COMPRESSABLE); + $payload = new Grpc\Testing\Payload(); + $payload->setType(Grpc\Testing\PayloadType::COMPRESSABLE); $payload->setBody(str_repeat("\0", $request_len)); $request->setPayload($payload); @@ -449,9 +453,9 @@ function customMetadata($stub) $streaming_call = $stub->FullDuplexCall($metadata); - $streaming_request = new grpc\testing\StreamingOutputCallRequest(); + $streaming_request = new Grpc\Testing\StreamingOutputCallRequest(); $streaming_request->setPayload($payload); - $response_parameters = new grpc\testing\ResponseParameters(); + $response_parameters = new Grpc\Testing\ResponseParameters(); $response_parameters->setSize($response_len); $streaming_request->getResponseParameters()[] = $response_parameters; $streaming_call->write($streaming_request); @@ -477,11 +481,11 @@ function customMetadata($stub) function statusCodeAndMessage($stub) { - $echo_status = new grpc\testing\EchoStatus(); + $echo_status = new Grpc\Testing\EchoStatus(); $echo_status->setCode(2); $echo_status->setMessage('test status message'); - $request = new grpc\testing\SimpleRequest(); + $request = new Grpc\Testing\SimpleRequest(); $request->setResponseStatus($echo_status); $call = $stub->UnaryCall($request); @@ -496,7 +500,7 @@ function statusCodeAndMessage($stub) $streaming_call = $stub->FullDuplexCall(); - $streaming_request = new grpc\testing\StreamingOutputCallRequest(); + $streaming_request = new Grpc\Testing\StreamingOutputCallRequest(); $streaming_request->setResponseStatus($echo_status); $streaming_call->write($streaming_request); $streaming_call->writesDone(); @@ -514,7 +518,7 @@ function statusCodeAndMessage($stub) # NOTE: the stub input to this function is from UnimplementedService function unimplementedService($stub) { - $call = $stub->UnimplementedCall(new grpc\testing\EmptyMessage()); + $call = $stub->UnimplementedCall(new Grpc\Testing\EmptyMessage()); list($result, $status) = $call->wait(); hardAssert($status->code === Grpc\STATUS_UNIMPLEMENTED, 'Received unexpected status code'); @@ -523,7 +527,7 @@ function unimplementedService($stub) # NOTE: the stub input to this function is from TestService function unimplementedMethod($stub) { - $call = $stub->UnimplementedCall(new grpc\testing\EmptyMessage()); + $call = $stub->UnimplementedCall(new Grpc\Testing\EmptyMessage()); list($result, $status) = $call->wait(); hardAssert($status->code === Grpc\STATUS_UNIMPLEMENTED, 'Received unexpected status code'); @@ -614,10 +618,10 @@ function _makeStub($args) } if ($test_case === 'unimplemented_service') { - $stub = new grpc\testing\UnimplementedServiceClient($server_address, + $stub = new Grpc\Testing\UnimplementedServiceClient($server_address, $opts); } else { - $stub = new grpc\testing\TestServiceClient($server_address, $opts); + $stub = new Grpc\Testing\TestServiceClient($server_address, $opts); } return $stub; diff --git a/src/proto/grpc/testing/BUILD b/src/proto/grpc/testing/BUILD index 283740839d..23a16a7cfc 100644 --- a/src/proto/grpc/testing/BUILD +++ b/src/proto/grpc/testing/BUILD @@ -42,11 +42,13 @@ grpc_proto_library( name = "control_proto", srcs = ["control.proto"], deps = ["payloads_proto", "stats_proto"], + has_services = False, ) grpc_proto_library( name = "echo_messages_proto", srcs = ["echo_messages.proto"], + has_services = False, ) grpc_proto_library( @@ -58,11 +60,13 @@ grpc_proto_library( grpc_proto_library( name = "empty_proto", srcs = ["empty.proto"], + has_services = False, ) grpc_proto_library( name = "messages_proto", srcs = ["messages.proto"], + has_services = False, ) grpc_proto_library( @@ -73,6 +77,7 @@ grpc_proto_library( grpc_proto_library( name = "payloads_proto", srcs = ["payloads.proto"], + has_services = False, ) grpc_proto_library( @@ -84,6 +89,7 @@ grpc_proto_library( grpc_proto_library( name = "stats_proto", srcs = ["stats.proto"], + has_services = False, ) grpc_proto_library( diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py index 5a8a3d487a..77412236cc 100644 --- a/src/python/grpcio/grpc/_channel.py +++ b/src/python/grpcio/grpc/_channel.py @@ -842,8 +842,8 @@ def _poll_connectivity(state, channel, initial_try_to_connect): connectivity = channel.check_connectivity_state(try_to_connect) with state.lock: state.connectivity = ( - _common. - CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[connectivity]) + _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[ + connectivity]) callbacks = tuple(callback for callback, unused_but_known_to_be_none_connectivity in state.callbacks_and_connectivities) diff --git a/src/python/grpcio/grpc/beta/_server_adaptations.py b/src/python/grpcio/grpc/beta/_server_adaptations.py index bb7c0960d5..206bd7e468 100644 --- a/src/python/grpcio/grpc/beta/_server_adaptations.py +++ b/src/python/grpcio/grpc/beta/_server_adaptations.py @@ -393,5 +393,4 @@ def server(service_implementations, multi_method_implementation, else: effective_thread_pool = thread_pool return _Server( - grpc.server( - effective_thread_pool, handlers=(generic_rpc_handler,))) + grpc.server(effective_thread_pool, handlers=(generic_rpc_handler,))) diff --git a/src/python/grpcio/grpc/framework/foundation/logging_pool.py b/src/python/grpcio/grpc/framework/foundation/logging_pool.py index 9164173d34..7ee37373fa 100644 --- a/src/python/grpcio/grpc/framework/foundation/logging_pool.py +++ b/src/python/grpcio/grpc/framework/foundation/logging_pool.py @@ -64,9 +64,8 @@ class _LoggingPool(object): return self._backing_pool.submit(_wrap(fn), *args, **kwargs) def map(self, func, *iterables, **kwargs): - return self._backing_pool.map(_wrap(func), - *iterables, - timeout=kwargs.get('timeout', None)) + return self._backing_pool.map( + _wrap(func), *iterables, timeout=kwargs.get('timeout', None)) def shutdown(self, wait=True): self._backing_pool.shutdown(wait=wait) diff --git a/src/python/grpcio_tests/tests/interop/methods.py b/src/python/grpcio_tests/tests/interop/methods.py index bdb258591e..1f9b356eb2 100644 --- a/src/python/grpcio_tests/tests/interop/methods.py +++ b/src/python/grpcio_tests/tests/interop/methods.py @@ -351,8 +351,7 @@ def _status_code_and_message(stub): response_type=messages_pb2.COMPRESSABLE, response_size=1, payload=messages_pb2.Payload(body=b'\x00'), - response_status=messages_pb2.EchoStatus( - code=code, message=details)) + response_status=messages_pb2.EchoStatus(code=code, message=details)) response_future = stub.UnaryCall.future(request) _validate_status_code_and_details(response_future, status, details) @@ -363,8 +362,7 @@ def _status_code_and_message(stub): response_type=messages_pb2.COMPRESSABLE, response_parameters=(messages_pb2.ResponseParameters(size=1),), payload=messages_pb2.Payload(body=b'\x00'), - response_status=messages_pb2.EchoStatus( - code=code, message=details)) + response_status=messages_pb2.EchoStatus(code=code, message=details)) pipe.add(request) # sends the initial request. # Dropping out of with block closes the pipe _validate_status_code_and_details(response_iterator, status, details) @@ -428,8 +426,8 @@ def _compute_engine_creds(stub, args): def _oauth2_auth_token(stub, args): - json_key_filename = os.environ[oauth2client_client. - GOOGLE_APPLICATION_CREDENTIALS] + json_key_filename = os.environ[ + oauth2client_client.GOOGLE_APPLICATION_CREDENTIALS] wanted_email = json.load(open(json_key_filename, 'rb'))['client_email'] response = _large_unary_common_behavior(stub, True, True, None) if wanted_email != response.username: @@ -441,8 +439,8 @@ def _oauth2_auth_token(stub, args): def _jwt_token_creds(stub, args): - json_key_filename = os.environ[oauth2client_client. - GOOGLE_APPLICATION_CREDENTIALS] + json_key_filename = os.environ[ + oauth2client_client.GOOGLE_APPLICATION_CREDENTIALS] wanted_email = json.load(open(json_key_filename, 'rb'))['client_email'] response = _large_unary_common_behavior(stub, True, False, None) if wanted_email != response.username: @@ -451,8 +449,8 @@ def _jwt_token_creds(stub, args): def _per_rpc_creds(stub, args): - json_key_filename = os.environ[oauth2client_client. - GOOGLE_APPLICATION_CREDENTIALS] + json_key_filename = os.environ[ + oauth2client_client.GOOGLE_APPLICATION_CREDENTIALS] wanted_email = json.load(open(json_key_filename, 'rb'))['client_email'] credentials = oauth2client_client.GoogleCredentials.get_application_default() scoped_credentials = credentials.create_scoped([args.oauth_scope]) diff --git a/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py b/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py index 0ca06868b2..5c7f903015 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py @@ -59,8 +59,7 @@ def _create_loop_destroy(): def _in_parallel(behavior, arguments): threads = tuple( - threading.Thread( - target=behavior, args=arguments) + threading.Thread(target=behavior, args=arguments) for _ in range(test_constants.THREAD_CONCURRENCY)) for thread in threads: thread.start() |