diff options
Diffstat (limited to 'src')
20 files changed, 527 insertions, 97 deletions
diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index 4cc6a4cf39..d3e71625d6 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -1408,4 +1408,180 @@ grpc::string GetSourceEpilogue(grpc_generator::File *file, return temp; } +// TODO(mmukhi): Make sure we need parameters or not. +grpc::string GetMockPrologue(grpc_generator::File *file, + const Parameters & /*params*/) { + grpc::string output; + { + // Scope the output stream so it closes and finalizes output to the string. + auto printer = file->CreatePrinter(&output); + std::map<grpc::string, grpc::string> vars; + + vars["filename"] = file->filename(); + vars["filename_base"] = file->filename_without_ext(); + vars["message_header_ext"] = message_header_ext(); + vars["service_header_ext"] = service_header_ext(); + + printer->Print(vars, "// Generated by the gRPC C++ plugin.\n"); + printer->Print(vars, + "// If you make any local change, they will be lost.\n"); + printer->Print(vars, "// source: $filename$\n\n"); + + printer->Print(vars, "#include \"$filename_base$$message_header_ext$\"\n"); + printer->Print(vars, "#include \"$filename_base$$service_header_ext$\"\n"); + printer->Print(vars, file->additional_headers().c_str()); + printer->Print(vars, "\n"); + } + return output; +} + +// TODO(mmukhi): Add client-stream and completion-queue headers. +grpc::string GetMockIncludes(grpc_generator::File *file, + const Parameters ¶ms) { + grpc::string output; + { + // Scope the output stream so it closes and finalizes output to the string. + auto printer = file->CreatePrinter(&output); + std::map<grpc::string, grpc::string> vars; + + static const char *headers_strs[] = { + "grpc++/impl/codegen/async_stream.h", + "grpc++/impl/codegen/sync_stream.h", "gmock/gmock.h", + }; + std::vector<grpc::string> headers(headers_strs, array_end(headers_strs)); + PrintIncludes(printer.get(), headers, params); + + if (!file->package().empty()) { + std::vector<grpc::string> parts = file->package_parts(); + + for (auto part = parts.begin(); part != parts.end(); part++) { + vars["part"] = *part; + printer->Print(vars, "namespace $part$ {\n"); + } + } + + printer->Print(vars, "\n"); + } + return output; +} + +void PrintMockClientMethods(grpc_generator::Printer *printer, + const grpc_generator::Method *method, + std::map<grpc::string, grpc::string> *vars) { + (*vars)["Method"] = method->name(); + (*vars)["Request"] = method->input_type_name(); + (*vars)["Response"] = method->output_type_name(); + + if (method->NoStreaming()) { + printer->Print( + *vars, + "MOCK_METHOD3($Method$, ::grpc::Status(::grpc::ClientContext* context, " + "const $Request$& request, $Response$* response));\n"); + printer->Print(*vars, + "MOCK_METHOD3(Async$Method$Raw, " + "::grpc::ClientAsyncResponseReaderInterface< $Response$>*" + "(::grpc::ClientContext* context, const $Request$& request, " + "::grpc::CompletionQueue* cq));\n"); + } else if (ClientOnlyStreaming(method)) { + printer->Print( + *vars, + "MOCK_METHOD2($Method$Raw, " + "::grpc::ClientWriterInterface< $Request$>*" + "(::grpc::ClientContext* context, $Response$* response));\n"); + printer->Print(*vars, + "MOCK_METHOD4(Async$Method$Raw, " + "::grpc::ClientAsyncWriterInterface< $Request$>*" + "(::grpc::ClientContext* context, $Response$* response, " + "::grpc::CompletionQueue* cq, void* tag));\n"); + } else if (ServerOnlyStreaming(method)) { + printer->Print( + *vars, + "MOCK_METHOD2($Method$Raw, " + "::grpc::ClientReaderInterface< $Response$>*" + "(::grpc::ClientContext* context, const $Request$& request));\n"); + printer->Print(*vars, + "MOCK_METHOD4(Async$Method$Raw, " + "::grpc::ClientAsyncReaderInterface< $Response$>*" + "(::grpc::ClientContext* context, const $Request$& request, " + "::grpc::CompletionQueue* cq, void* tag));\n"); + } else if (method->BidiStreaming()) { + printer->Print( + *vars, + "MOCK_METHOD1($Method$Raw, " + "::grpc::ClientReaderWriterInterface< $Request$, $Response$>*" + "(::grpc::ClientContext* context));\n"); + printer->Print( + *vars, + "MOCK_METHOD3(Async$Method$Raw, " + "::grpc::ClientAsyncReaderWriterInterface<$Request$, $Response$>*" + "(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, " + "void* tag));\n"); + } +} + +void PrintMockService(grpc_generator::Printer *printer, + const grpc_generator::Service *service, + std::map<grpc::string, grpc::string> *vars) { + (*vars)["Service"] = service->name(); + + printer->Print(*vars, + "class Mock$Service$Stub : public $Service$::StubInterface {\n" + " public:\n"); + printer->Indent(); + for (int i = 0; i < service->method_count(); ++i) { + PrintMockClientMethods(printer, service->method(i).get(), vars); + } + printer->Outdent(); + printer->Print("};\n"); +} + +grpc::string GetMockServices(grpc_generator::File *file, + const Parameters ¶ms) { + grpc::string output; + { + // Scope the output stream so it closes and finalizes output to the string. + auto printer = file->CreatePrinter(&output); + std::map<grpc::string, grpc::string> vars; + // Package string is empty or ends with a dot. It is used to fully qualify + // method names. + vars["Package"] = file->package(); + if (!file->package().empty()) { + vars["Package"].append("."); + } + + if (!params.services_namespace.empty()) { + vars["services_namespace"] = params.services_namespace; + printer->Print(vars, "\nnamespace $services_namespace$ {\n\n"); + } + + for (int i = 0; i < file->service_count(); i++) { + PrintMockService(printer.get(), file->service(i).get(), &vars); + printer->Print("\n"); + } + + if (!params.services_namespace.empty()) { + printer->Print(vars, "} // namespace $services_namespace$\n\n"); + } + } + return output; +} + +grpc::string GetMockEpilogue(grpc_generator::File *file, + const Parameters & /*params*/) { + grpc::string temp; + + if (!file->package().empty()) { + std::vector<grpc::string> parts = file->package_parts(); + + for (auto part = parts.begin(); part != parts.end(); part++) { + temp.append("} // namespace "); + temp.append(*part); + temp.append("\n"); + } + temp.append("\n"); + } + + return temp; +} + } // namespace grpc_cpp_generator diff --git a/src/compiler/cpp_generator.h b/src/compiler/cpp_generator.h index 69fd8a93e9..6119ebe289 100644 --- a/src/compiler/cpp_generator.h +++ b/src/compiler/cpp_generator.h @@ -65,6 +65,8 @@ struct Parameters { bool use_system_headers; // Prefix to any grpc include grpc::string grpc_search_path; + // Generate GMOCK code to facilitate unit testing. + bool generate_mock_code; }; // Return the prologue of the generated header file. @@ -99,6 +101,38 @@ grpc::string GetSourceServices(grpc_generator::File *file, grpc::string GetSourceEpilogue(grpc_generator::File *file, const Parameters ¶ms); +// Return the prologue of the generated mock file. +grpc::string GetMockPrologue(grpc_generator::File *file, + const Parameters ¶ms); + +// Return the includes needed for generated mock file. +grpc::string GetMockIncludes(grpc_generator::File *file, + const Parameters ¶ms); + +// Return the services for generated mock file. +grpc::string GetMockServices(grpc_generator::File *file, + const Parameters ¶ms); + +// Return the epilogue of generated mock file. +grpc::string GetMockEpilogue(grpc_generator::File *file, + const Parameters ¶ms); + +// Return the prologue of the generated mock file. +grpc::string GetMockPrologue(grpc_generator::File *file, + const Parameters ¶ms); + +// Return the includes needed for generated mock file. +grpc::string GetMockIncludes(grpc_generator::File *file, + const Parameters ¶ms); + +// Return the services for generated mock file. +grpc::string GetMockServices(grpc_generator::File *file, + const Parameters ¶ms); + +// Return the epilogue of generated mock file. +grpc::string GetMockEpilogue(grpc_generator::File *file, + const Parameters ¶ms); + } // namespace grpc_cpp_generator #endif // GRPC_INTERNAL_COMPILER_CPP_GENERATOR_H diff --git a/src/compiler/cpp_plugin.cc b/src/compiler/cpp_plugin.cc index 4ee05ee037..35f1bf3e93 100644 --- a/src/compiler/cpp_plugin.cc +++ b/src/compiler/cpp_plugin.cc @@ -62,6 +62,7 @@ class CppGrpcGenerator : public grpc::protobuf::compiler::CodeGenerator { grpc_cpp_generator::Parameters generator_parameters; generator_parameters.use_system_headers = true; + generator_parameters.generate_mock_code = false; ProtoBufFile pbfile(file); @@ -85,6 +86,13 @@ class CppGrpcGenerator : public grpc::protobuf::compiler::CodeGenerator { } } else if (param[0] == "grpc_search_path") { generator_parameters.grpc_search_path = param[1]; + } else if (param[0] == "generate_mock_code") { + if (param[1] == "true") { + generator_parameters.generate_mock_code = true; + } else if (param[1] != "false") { + *error = grpc::string("Invalid parameter: ") + *parameter_string; + return false; + } } else { *error = grpc::string("Unknown parameter: ") + *parameter_string; return false; @@ -114,6 +122,19 @@ class CppGrpcGenerator : public grpc::protobuf::compiler::CodeGenerator { grpc::protobuf::io::CodedOutputStream source_coded_out(source_output.get()); source_coded_out.WriteRaw(source_code.data(), source_code.size()); + if (!generator_parameters.generate_mock_code) { + return true; + } + grpc::string mock_code = + grpc_cpp_generator::GetMockPrologue(&pbfile, generator_parameters) + + grpc_cpp_generator::GetMockIncludes(&pbfile, generator_parameters) + + grpc_cpp_generator::GetMockServices(&pbfile, generator_parameters) + + grpc_cpp_generator::GetMockEpilogue(&pbfile, generator_parameters); + std::unique_ptr<grpc::protobuf::io::ZeroCopyOutputStream> mock_output( + context->Open(file_name + "_mock.grpc.pb.h")); + grpc::protobuf::io::CodedOutputStream mock_coded_out(mock_output.get()); + mock_coded_out.WriteRaw(mock_code.data(), mock_code.size()); + return true; } diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c index 8d28e829d8..0463b25412 100644 --- a/src/core/ext/filters/client_channel/client_channel.c +++ b/src/core/ext/filters/client_channel/client_channel.c @@ -400,26 +400,24 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, GPR_ASSERT(channel_arg->type == GRPC_ARG_STRING); lb_policy_name = channel_arg->value.string; } - // Special case: If all of the addresses are balancer addresses, - // assume that we should use the grpclb policy, regardless of what the - // resolver actually specified. + // Special case: If at least one balancer address is present, we use + // the grpclb policy, regardless of what the resolver actually specified. channel_arg = grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_ADDRESSES); if (channel_arg != NULL && channel_arg->type == GRPC_ARG_POINTER) { grpc_lb_addresses *addresses = channel_arg->value.pointer.p; - bool found_backend_address = false; + bool found_balancer_address = false; for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (!addresses->addresses[i].is_balancer) { - found_backend_address = true; + if (addresses->addresses[i].is_balancer) { + found_balancer_address = true; break; } } - if (!found_backend_address) { + if (found_balancer_address) { if (lb_policy_name != NULL && strcmp(lb_policy_name, "grpclb") != 0) { gpr_log(GPR_INFO, - "resolver requested LB policy %s but provided only balancer " - "addresses, no backend addresses -- forcing use of grpclb LB " - "policy", + "resolver requested LB policy %s but provided at least one " + "balancer address -- forcing use of grpclb LB policy", lb_policy_name); } lb_policy_name = "grpclb"; diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c index f8524732df..ad5f0685ec 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c @@ -831,10 +831,10 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, /* Count the number of gRPC-LB addresses. There must be at least one. * TODO(roth): For now, we ignore non-balancer addresses, but in the * future, we may change the behavior such that we fall back to using - * the non-balancer addresses if we cannot reach any balancers. At that - * time, this should be changed to allow a list with no balancer addresses, - * since the resolver might fail to return a balancer address even when - * this is the right LB policy to use. */ + * the non-balancer addresses if we cannot reach any balancers. In the + * fallback case, we should use the LB policy indicated by + * GRPC_ARG_LB_POLICY_NAME (although if that specifies grpclb or is + * unset, we should default to pick_first). */ const grpc_arg *arg = grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES); if (arg == NULL || arg->type != GRPC_ARG_POINTER) { diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.c b/src/core/ext/transport/chttp2/server/chttp2_server.c index b463506a98..b9c62c376a 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.c +++ b/src/core/ext/transport/chttp2/server/chttp2_server.c @@ -80,7 +80,7 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, gpr_mu_lock(&connection_state->server_state->mu); if (error != GRPC_ERROR_NONE || connection_state->server_state->shutdown) { const char *error_str = grpc_error_string(error); - gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str); + gpr_log(GPR_DEBUG, "Handshaking failed: %s", error_str); if (error == GRPC_ERROR_NONE && args->endpoint != NULL) { // We were shut down after handshaking completed successfully, so diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index ca283d034f..af70746064 100644 --- a/src/core/lib/iomgr/udp_server.c +++ b/src/core/lib/iomgr/udp_server.c @@ -92,6 +92,11 @@ struct grpc_udp_listener { struct grpc_udp_listener *next; }; +struct shutdown_fd_args { + grpc_fd *fd; + gpr_mu *server_mu; +}; + /* the overall server */ struct grpc_udp_server { gpr_mu mu; @@ -151,8 +156,13 @@ grpc_udp_server *grpc_udp_server_create(const grpc_channel_args *args) { return s; } -static void shutdown_fd(grpc_exec_ctx *exec_ctx, void *fd, grpc_error *error) { - grpc_fd_shutdown(exec_ctx, (grpc_fd *)fd, GRPC_ERROR_REF(error)); +static void shutdown_fd(grpc_exec_ctx *exec_ctx, void *args, + grpc_error *error) { + struct shutdown_fd_args *shutdown_args = (struct shutdown_fd_args *)args; + gpr_mu_lock(shutdown_args->server_mu); + grpc_fd_shutdown(exec_ctx, shutdown_args->fd, GRPC_ERROR_REF(error)); + gpr_mu_unlock(shutdown_args->server_mu); + gpr_free(shutdown_args); } static void dummy_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { @@ -242,7 +252,10 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, if (s->active_ports) { for (sp = s->head; sp; sp = sp->next) { GPR_ASSERT(sp->orphan_cb); - grpc_closure_init(&sp->orphan_fd_closure, shutdown_fd, sp->emfd, + struct shutdown_fd_args *args = gpr_malloc(sizeof(*args)); + args->fd = sp->emfd; + args->server_mu = &s->mu; + grpc_closure_init(&sp->orphan_fd_closure, shutdown_fd, args, grpc_schedule_on_exec_ctx); sp->orphan_cb(exec_ctx, sp->emfd, &sp->orphan_fd_closure, sp->server->user_data); diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index c4ee222043..122156e93c 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -61,6 +61,146 @@ typedef struct { void *tag; } plucker; +typedef struct { + bool can_get_pollset; + bool can_listen; + size_t (*size)(void); + void (*init)(grpc_pollset *pollset, gpr_mu **mu); + grpc_error *(*kick)(grpc_pollset *pollset, + grpc_pollset_worker *specific_worker); + grpc_error *(*work)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, + grpc_pollset_worker **worker, gpr_timespec now, + gpr_timespec deadline); + void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, + grpc_closure *closure); + void (*destroy)(grpc_pollset *pollset); +} cq_poller_vtable; + +typedef struct non_polling_worker { + gpr_cv cv; + bool kicked; + struct non_polling_worker *next; + struct non_polling_worker *prev; +} non_polling_worker; + +typedef struct { + gpr_mu mu; + non_polling_worker *root; + grpc_closure *shutdown; +} non_polling_poller; + +static size_t non_polling_poller_size(void) { + return sizeof(non_polling_poller); +} + +static void non_polling_poller_init(grpc_pollset *pollset, gpr_mu **mu) { + non_polling_poller *npp = (non_polling_poller *)pollset; + gpr_mu_init(&npp->mu); + *mu = &npp->mu; +} + +static void non_polling_poller_destroy(grpc_pollset *pollset) { + non_polling_poller *npp = (non_polling_poller *)pollset; + gpr_mu_destroy(&npp->mu); +} + +static grpc_error *non_polling_poller_work(grpc_exec_ctx *exec_ctx, + grpc_pollset *pollset, + grpc_pollset_worker **worker, + gpr_timespec now, + gpr_timespec deadline) { + non_polling_poller *npp = (non_polling_poller *)pollset; + if (npp->shutdown) return GRPC_ERROR_NONE; + non_polling_worker w; + gpr_cv_init(&w.cv); + if (worker != NULL) *worker = (grpc_pollset_worker *)&w; + if (npp->root == NULL) { + npp->root = w.next = w.prev = &w; + } else { + w.next = npp->root; + w.prev = w.next->prev; + w.next->prev = w.prev->next = &w; + } + w.kicked = false; + while (!npp->shutdown && !w.kicked && !gpr_cv_wait(&w.cv, &npp->mu, deadline)) + ; + if (&w == npp->root) { + npp->root = w.next; + if (&w == npp->root) { + if (npp->shutdown) { + grpc_closure_sched(exec_ctx, npp->shutdown, GRPC_ERROR_NONE); + } + npp->root = NULL; + } + } + w.next->prev = w.prev; + w.prev->next = w.next; + gpr_cv_destroy(&w.cv); + if (worker != NULL) *worker = NULL; + return GRPC_ERROR_NONE; +} + +static grpc_error *non_polling_poller_kick( + grpc_pollset *pollset, grpc_pollset_worker *specific_worker) { + non_polling_poller *p = (non_polling_poller *)pollset; + if (specific_worker == NULL) specific_worker = (grpc_pollset_worker *)p->root; + if (specific_worker != NULL) { + non_polling_worker *w = (non_polling_worker *)specific_worker; + if (!w->kicked) { + w->kicked = true; + gpr_cv_signal(&w->cv); + } + } + return GRPC_ERROR_NONE; +} + +static void non_polling_poller_shutdown(grpc_exec_ctx *exec_ctx, + grpc_pollset *pollset, + grpc_closure *closure) { + non_polling_poller *p = (non_polling_poller *)pollset; + GPR_ASSERT(closure != NULL); + p->shutdown = closure; + if (p->root == NULL) { + grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE); + } else { + non_polling_worker *w = p->root; + do { + gpr_cv_signal(&w->cv); + w = w->next; + } while (w != p->root); + } +} + +static const cq_poller_vtable g_poller_vtable_by_poller_type[] = { + /* GRPC_CQ_DEFAULT_POLLING */ + {.can_get_pollset = true, + .can_listen = true, + .size = grpc_pollset_size, + .init = grpc_pollset_init, + .kick = grpc_pollset_kick, + .work = grpc_pollset_work, + .shutdown = grpc_pollset_shutdown, + .destroy = grpc_pollset_destroy}, + /* GRPC_CQ_NON_LISTENING */ + {.can_get_pollset = true, + .can_listen = false, + .size = grpc_pollset_size, + .init = grpc_pollset_init, + .kick = grpc_pollset_kick, + .work = grpc_pollset_work, + .shutdown = grpc_pollset_shutdown, + .destroy = grpc_pollset_destroy}, + /* GRPC_CQ_NON_POLLING */ + {.can_get_pollset = false, + .can_listen = false, + .size = non_polling_poller_size, + .init = non_polling_poller_init, + .kick = non_polling_poller_kick, + .work = non_polling_poller_work, + .shutdown = non_polling_poller_shutdown, + .destroy = non_polling_poller_destroy}, +}; + /* Queue that holds the cq_completion_events. This internally uses gpr_mpscq * queue (a lockfree multiproducer single consumer queue). However this queue * supports multiple consumers too. As such, it uses the queue_mu to serialize @@ -86,10 +226,10 @@ struct grpc_completion_queue { gpr_mu *mu; grpc_cq_completion_type completion_type; - grpc_cq_polling_type polling_type; - /** Completed events (Only relevant if the completion_type is NOT - * GRPC_CQ_NEXT) */ + const cq_poller_vtable *poller_vtable; + + /** completed events */ grpc_cq_completion completed_head; grpc_cq_completion *completed_tail; @@ -188,15 +328,18 @@ grpc_completion_queue *grpc_completion_queue_create_internal( "polling_type=%d)", 2, (completion_type, polling_type)); - cc = gpr_zalloc(sizeof(grpc_completion_queue) + grpc_pollset_size()); - grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu); + const cq_poller_vtable *poller_vtable = + &g_poller_vtable_by_poller_type[polling_type]; + + cc = gpr_zalloc(sizeof(grpc_completion_queue) + poller_vtable->size()); + poller_vtable->init(POLLSET_FROM_CQ(cc), &cc->mu); #ifndef NDEBUG cc->outstanding_tags = NULL; cc->outstanding_tag_capacity = 0; #endif cc->completion_type = completion_type; - cc->polling_type = polling_type; + cc->poller_vtable = poller_vtable; /* Initial ref is dropped by grpc_completion_queue_shutdown */ gpr_ref_init(&cc->pending_events, 1); @@ -257,7 +400,7 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) { #endif if (gpr_unref(&cc->owning_refs)) { GPR_ASSERT(cc->completed_head.next == (uintptr_t)&cc->completed_head); - grpc_pollset_destroy(POLLSET_FROM_CQ(cc)); + cc->poller_vtable->destroy(POLLSET_FROM_CQ(cc)); cq_event_queue_destroy(&cc->queue); #ifndef NDEBUG gpr_free(cc->outstanding_tags); @@ -331,7 +474,8 @@ void grpc_cq_end_op_for_next(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, int shutdown = gpr_unref(&cc->pending_events); if (!shutdown) { gpr_mu_lock(cc->mu); - grpc_error *kick_error = grpc_pollset_kick(POLLSET_FROM_CQ(cc), NULL); + + grpc_error *kick_error = cc->poller_vtable->kick(POLLSET_FROM_CQ(cc), NULL); gpr_mu_unlock(cc->mu); if (kick_error != GRPC_ERROR_NONE) { @@ -347,8 +491,8 @@ void grpc_cq_end_op_for_next(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, gpr_atm_no_barrier_store(&cc->shutdown, 1); gpr_mu_lock(cc->mu); - grpc_pollset_shutdown(exec_ctx, POLLSET_FROM_CQ(cc), - &cc->pollset_shutdown_done); + cc->poller_vtable->shutdown(exec_ctx, POLLSET_FROM_CQ(cc), + &cc->pollset_shutdown_done); gpr_mu_unlock(cc->mu); } } @@ -387,7 +531,8 @@ void grpc_cq_end_op_for_pluck(grpc_exec_ctx *exec_ctx, } grpc_error *kick_error = - grpc_pollset_kick(POLLSET_FROM_CQ(cc), pluck_worker); + cc->poller_vtable->kick(POLLSET_FROM_CQ(cc), pluck_worker); + gpr_mu_unlock(cc->mu); if (kick_error != GRPC_ERROR_NONE) { @@ -400,8 +545,8 @@ void grpc_cq_end_op_for_pluck(grpc_exec_ctx *exec_ctx, GPR_ASSERT(!gpr_atm_no_barrier_load(&cc->shutdown)); GPR_ASSERT(cc->shutdown_called); gpr_atm_no_barrier_store(&cc->shutdown, 1); - grpc_pollset_shutdown(exec_ctx, POLLSET_FROM_CQ(cc), - &cc->pollset_shutdown_done); + cc->poller_vtable->shutdown(exec_ctx, POLLSET_FROM_CQ(cc), + &cc->pollset_shutdown_done); gpr_mu_unlock(cc->mu); } } @@ -477,7 +622,6 @@ static bool cq_is_next_finished(grpc_exec_ctx *exec_ctx, void *arg) { return true; } } - return !a->first_loop && gpr_time_cmp(a->deadline, gpr_now(a->deadline.clock_type)) < 0; } @@ -534,6 +678,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); GRPC_CQ_INTERNAL_REF(cc, "next"); + cq_is_finished_arg is_finished_arg = { .last_seen_things_queued_ever = gpr_atm_no_barrier_load(&cc->things_queued_ever), @@ -601,9 +746,9 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, dump_pending_tags(cc); break; } - - /* Check alarms - these are a global resource so we just ping each time - through on every pollset. May update deadline to ensure timely wakeups.*/ + /* Check alarms - these are a global resource so we just ping + each time through on every pollset. + May update deadline to ensure timely wakeups. if (grpc_timer_check(&exec_ctx, now, &iteration_deadline)) { GPR_TIMER_MARK("alarm_triggered", 0); grpc_exec_ctx_flush(&exec_ctx); @@ -612,8 +757,8 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, /* The main polling work happens in grpc_pollset_work */ gpr_mu_lock(cc->mu); - grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), NULL, - now, iteration_deadline); + grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc), + NULL, now, iteration_deadline); gpr_mu_unlock(cc->mu); if (err != GRPC_ERROR_NONE) { @@ -805,8 +950,8 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, grpc_exec_ctx_flush(&exec_ctx); gpr_mu_lock(cc->mu); } else { - grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), - &worker, now, iteration_deadline); + grpc_error *err = cc->poller_vtable->work( + &exec_ctx, POLLSET_FROM_CQ(cc), &worker, now, iteration_deadline); if (err != GRPC_ERROR_NONE) { del_plucker(cc, tag, &worker); gpr_mu_unlock(cc->mu); @@ -850,8 +995,8 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) { if (gpr_unref(&cc->pending_events)) { GPR_ASSERT(!cc->shutdown); cc->shutdown = 1; - grpc_pollset_shutdown(&exec_ctx, POLLSET_FROM_CQ(cc), - &cc->pollset_shutdown_done); + cc->poller_vtable->shutdown(&exec_ctx, POLLSET_FROM_CQ(cc), + &cc->pollset_shutdown_done); } gpr_mu_unlock(cc->mu); grpc_exec_ctx_finish(&exec_ctx); @@ -872,7 +1017,7 @@ void grpc_completion_queue_destroy(grpc_completion_queue *cc) { } grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) { - return POLLSET_FROM_CQ(cc); + return cc->poller_vtable->can_get_pollset ? POLLSET_FROM_CQ(cc) : NULL; } grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps) { @@ -893,4 +1038,10 @@ bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc) { void grpc_cq_mark_server_cq(grpc_completion_queue *cc) { cc->is_server_cq = 1; } -int grpc_cq_is_server_cq(grpc_completion_queue *cc) { return cc->is_server_cq; } +bool grpc_cq_is_server_cq(grpc_completion_queue *cc) { + return cc->is_server_cq; +} + +bool grpc_cq_can_listen(grpc_completion_queue *cc) { + return cc->poller_vtable->can_listen; +} diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 9c8bc3b53a..f7eb148982 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -96,13 +96,11 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc); grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps); -void grpc_cq_mark_non_listening_server_cq(grpc_completion_queue *cc); -bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc); void grpc_cq_mark_server_cq(grpc_completion_queue *cc); -int grpc_cq_is_server_cq(grpc_completion_queue *cc); +bool grpc_cq_is_server_cq(grpc_completion_queue *cc); +bool grpc_cq_can_listen(grpc_completion_queue *cc); grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc); -grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc); grpc_completion_queue *grpc_completion_queue_create_internal( grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type); diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index e133d1d2a4..934ca0431a 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -981,7 +981,7 @@ const grpc_channel_filter grpc_server_top_filter = { static void register_completion_queue(grpc_server *server, grpc_completion_queue *cq, - bool is_non_listening, void *reserved) { + void *reserved) { size_t i, n; GPR_ASSERT(!reserved); for (i = 0; i < server->cq_count; i++) { @@ -990,10 +990,6 @@ static void register_completion_queue(grpc_server *server, grpc_cq_mark_server_cq(cq); - if (is_non_listening) { - grpc_cq_mark_non_listening_server_cq(cq); - } - GRPC_CQ_INTERNAL_REF(cq, "server"); n = server->cq_count++; server->cqs = gpr_realloc(server->cqs, @@ -1016,16 +1012,7 @@ void grpc_server_register_completion_queue(grpc_server *server, calls grpc_completion_queue_pluck() on server completion queues */ } - register_completion_queue(server, cq, false, reserved); -} - -void grpc_server_register_non_listening_completion_queue( - grpc_server *server, grpc_completion_queue *cq, void *reserved) { - GRPC_API_TRACE( - "grpc_server_register_non_listening_completion_queue(server=%p, cq=%p, " - "reserved=%p)", - 3, (server, cq, reserved)); - register_completion_queue(server, cq, true, reserved); + register_completion_queue(server, cq, reserved); } grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) { @@ -1121,7 +1108,7 @@ void grpc_server_start(grpc_server *server) { server->requested_calls_per_cq = gpr_malloc(sizeof(*server->requested_calls_per_cq) * server->cq_count); for (i = 0; i < server->cq_count; i++) { - if (!grpc_cq_is_non_listening_server_cq(server->cqs[i])) { + if (grpc_cq_can_listen(server->cqs[i])) { server->pollsets[server->pollset_count++] = grpc_cq_pollset(server->cqs[i]); } diff --git a/src/cpp/common/core_codegen.cc b/src/cpp/common/core_codegen.cc index 4f8b826aec..f679a33945 100644 --- a/src/cpp/common/core_codegen.cc +++ b/src/cpp/common/core_codegen.cc @@ -54,6 +54,18 @@ struct grpc_byte_buffer; namespace grpc { +const grpc_completion_queue_factory* +CoreCodegen::grpc_completion_queue_factory_lookup( + const grpc_completion_queue_attributes* attributes) { + return ::grpc_completion_queue_factory_lookup(attributes); +} + +grpc_completion_queue* CoreCodegen::grpc_completion_queue_create( + const grpc_completion_queue_factory* factory, + const grpc_completion_queue_attributes* attributes, void* reserved) { + return ::grpc_completion_queue_create(factory, attributes, reserved); +} + grpc_completion_queue* CoreCodegen::grpc_completion_queue_create_for_next( void* reserved) { return ::grpc_completion_queue_create_for_next(reserved); diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index 3a15afc49e..2ead048a1f 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -83,7 +83,8 @@ ServerBuilder::~ServerBuilder() { std::unique_ptr<ServerCompletionQueue> ServerBuilder::AddCompletionQueue( bool is_frequently_polled) { - ServerCompletionQueue* cq = new ServerCompletionQueue(is_frequently_polled); + ServerCompletionQueue* cq = new ServerCompletionQueue( + is_frequently_polled ? GRPC_CQ_DEFAULT_POLLING : GRPC_CQ_NON_LISTENING); cqs_.push_back(cq); return std::unique_ptr<ServerCompletionQueue>(cq); } @@ -242,6 +243,16 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() { sync_server_cqs(std::make_shared< std::vector<std::unique_ptr<ServerCompletionQueue>>>()); + int num_frequently_polled_cqs = 0; + for (auto it = cqs_.begin(); it != cqs_.end(); ++it) { + if ((*it)->IsFrequentlyPolled()) { + num_frequently_polled_cqs++; + } + } + + const bool is_hybrid_server = + has_sync_methods && num_frequently_polled_cqs > 0; + if (has_sync_methods) { // This is a Sync server gpr_log(GPR_INFO, @@ -251,9 +262,12 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() { sync_server_settings_.max_pollers, sync_server_settings_.cq_timeout_msec); + grpc_cq_polling_type polling_type = + is_hybrid_server ? GRPC_CQ_NON_POLLING : GRPC_CQ_DEFAULT_POLLING; + // Create completion queues to listen to incoming rpc requests for (int i = 0; i < sync_server_settings_.num_cqs; i++) { - sync_server_cqs->emplace_back(new ServerCompletionQueue()); + sync_server_cqs->emplace_back(new ServerCompletionQueue(polling_type)); } } @@ -269,12 +283,10 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() { // server // 2. cqs_: Completion queues added via AddCompletionQueue() call - // All sync cqs (if any) are frequently polled by ThreadManager - int num_frequently_polled_cqs = sync_server_cqs->size(); - for (auto it = sync_server_cqs->begin(); it != sync_server_cqs->end(); ++it) { grpc_server_register_completion_queue(server->server_, (*it)->cq(), nullptr); + num_frequently_polled_cqs++; } // cqs_ contains the completion queue added by calling the ServerBuilder's @@ -283,14 +295,8 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() { // listening to incoming channels. Such completion queues must be registered // as non-listening queues for (auto it = cqs_.begin(); it != cqs_.end(); ++it) { - if ((*it)->IsFrequentlyPolled()) { - grpc_server_register_completion_queue(server->server_, (*it)->cq(), - nullptr); - num_frequently_polled_cqs++; - } else { - grpc_server_register_non_listening_completion_queue(server->server_, - (*it)->cq(), nullptr); - } + grpc_server_register_completion_queue(server->server_, (*it)->cq(), + nullptr); } if (num_frequently_polled_cqs == 0) { diff --git a/src/node/ext/server_generic.cc b/src/node/ext/server_generic.cc index 24573bd52f..088273d527 100644 --- a/src/node/ext/server_generic.cc +++ b/src/node/ext/server_generic.cc @@ -44,9 +44,11 @@ namespace grpc { namespace node { Server::Server(grpc_server *server) : wrapped_server(server) { - shutdown_queue = grpc_completion_queue_create_for_pluck(NULL); - grpc_server_register_non_listening_completion_queue(server, shutdown_queue, - NULL); + grpc_completion_queue_attributes attrs = { + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_NON_LISTENING}; + shutdown_queue = grpc_completion_queue_create( + grpc_completion_queue_factory_lookup(&attrs), &attrs, NULL); + grpc_server_register_completion_queue(server, shutdown_queue, NULL); } Server::~Server() { diff --git a/src/proto/grpc/health/v1/BUILD b/src/proto/grpc/health/v1/BUILD new file mode 100644 index 0000000000..dbb91d9139 --- /dev/null +++ b/src/proto/grpc/health/v1/BUILD @@ -0,0 +1,39 @@ +# Copyright 2017, Google Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following disclaimer +# in the documentation and/or other materials provided with the +# distribution. +# * Neither the name of Google Inc. nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +licenses(["notice"]) # 3-clause BSD + +package(default_visibility = ["//visibility:public"]) + +load("//bazel:grpc_build_system.bzl", "grpc_proto_library") + +grpc_proto_library( + name = "health_proto", + srcs = ["health.proto"], +) diff --git a/src/proto/grpc/testing/BUILD b/src/proto/grpc/testing/BUILD index 6f3422e4d1..805988c337 100644 --- a/src/proto/grpc/testing/BUILD +++ b/src/proto/grpc/testing/BUILD @@ -36,6 +36,7 @@ load("//bazel:grpc_build_system.bzl", "grpc_proto_library") grpc_proto_library( name = "compiler_test_proto", srcs = ["compiler_test.proto"], + generate_mock = True, ) grpc_proto_library( @@ -55,6 +56,7 @@ grpc_proto_library( name = "echo_proto", srcs = ["echo.proto"], deps = ["echo_messages_proto"], + generate_mock = True, ) grpc_proto_library( diff --git a/src/proto/grpc/testing/compiler_test.proto b/src/proto/grpc/testing/compiler_test.proto index 085e8ae59f..24735522e0 100644 --- a/src/proto/grpc/testing/compiler_test.proto +++ b/src/proto/grpc/testing/compiler_test.proto @@ -59,6 +59,14 @@ service ServiceA { // Method A2 leading comment 2 rpc MethodA2(stream Request) returns (Response); // MethodA2 trailing comment 1 + + // Method A3 leading comment 1 + rpc MethodA3(Request) returns (stream Response); + // Method A3 trailing comment 1 + + // Method A4 leading comment 1 + rpc MethodA4(stream Request) returns (stream Response); + // Method A4 trailing comment 1 } // Ignored ServiceA trailing comment 1 diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi index 4f1776e002..f66f6e4122 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi @@ -356,8 +356,6 @@ cdef extern from "grpc/grpc.h": void grpc_server_register_completion_queue(grpc_server *server, grpc_completion_queue *cq, void *reserved) nogil - void grpc_server_register_non_listening_completion_queue( - grpc_server *server, grpc_completion_queue *cq, void *reserved) nogil int grpc_server_add_insecure_http2_port( grpc_server *server, const char *addr) nogil void grpc_server_start(grpc_server *server) nogil @@ -502,4 +500,3 @@ cdef extern from "grpc/compression.h": int grpc_compression_options_is_algorithm_enabled( const grpc_compression_options *opts, grpc_compression_algorithm algorithm) nogil - diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi index 18db38b686..97192efda7 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi @@ -82,20 +82,11 @@ cdef class Server: self.c_server, queue.c_completion_queue, NULL) self.registered_completion_queues.append(queue) - def register_non_listening_completion_queue( - self, CompletionQueue queue not None): - if self.is_started: - raise ValueError("cannot register completion queues after start") - with nogil: - grpc_server_register_non_listening_completion_queue( - self.c_server, queue.c_completion_queue, NULL) - self.registered_completion_queues.append(queue) - def start(self): if self.is_started: raise ValueError("the server has already started") self.backup_shutdown_queue = CompletionQueue() - self.register_non_listening_completion_queue(self.backup_shutdown_queue) + self.register_completion_queue(self.backup_shutdown_queue) self.is_started = True with nogil: grpc_server_start(self.c_server) diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.c b/src/ruby/ext/grpc/rb_grpc_imports.generated.c index 1c68b2dc7c..a412277f6d 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.c +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.c @@ -127,7 +127,6 @@ grpc_server_register_method_type grpc_server_register_method_import; grpc_server_request_registered_call_type grpc_server_request_registered_call_import; grpc_server_create_type grpc_server_create_import; grpc_server_register_completion_queue_type grpc_server_register_completion_queue_import; -grpc_server_register_non_listening_completion_queue_type grpc_server_register_non_listening_completion_queue_import; grpc_server_add_insecure_http2_port_type grpc_server_add_insecure_http2_port_import; grpc_server_start_type grpc_server_start_import; grpc_server_shutdown_and_notify_type grpc_server_shutdown_and_notify_import; @@ -425,7 +424,6 @@ void grpc_rb_load_imports(HMODULE library) { grpc_server_request_registered_call_import = (grpc_server_request_registered_call_type) GetProcAddress(library, "grpc_server_request_registered_call"); grpc_server_create_import = (grpc_server_create_type) GetProcAddress(library, "grpc_server_create"); grpc_server_register_completion_queue_import = (grpc_server_register_completion_queue_type) GetProcAddress(library, "grpc_server_register_completion_queue"); - grpc_server_register_non_listening_completion_queue_import = (grpc_server_register_non_listening_completion_queue_type) GetProcAddress(library, "grpc_server_register_non_listening_completion_queue"); grpc_server_add_insecure_http2_port_import = (grpc_server_add_insecure_http2_port_type) GetProcAddress(library, "grpc_server_add_insecure_http2_port"); grpc_server_start_import = (grpc_server_start_type) GetProcAddress(library, "grpc_server_start"); grpc_server_shutdown_and_notify_import = (grpc_server_shutdown_and_notify_type) GetProcAddress(library, "grpc_server_shutdown_and_notify"); diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h index c33fdd210e..7df8dd69fc 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h @@ -332,9 +332,6 @@ extern grpc_server_create_type grpc_server_create_import; typedef void(*grpc_server_register_completion_queue_type)(grpc_server *server, grpc_completion_queue *cq, void *reserved); extern grpc_server_register_completion_queue_type grpc_server_register_completion_queue_import; #define grpc_server_register_completion_queue grpc_server_register_completion_queue_import -typedef void(*grpc_server_register_non_listening_completion_queue_type)(grpc_server *server, grpc_completion_queue *q, void *reserved); -extern grpc_server_register_non_listening_completion_queue_type grpc_server_register_non_listening_completion_queue_import; -#define grpc_server_register_non_listening_completion_queue grpc_server_register_non_listening_completion_queue_import typedef int(*grpc_server_add_insecure_http2_port_type)(grpc_server *server, const char *addr); extern grpc_server_add_insecure_http2_port_type grpc_server_add_insecure_http2_port_import; #define grpc_server_add_insecure_http2_port grpc_server_add_insecure_http2_port_import |