From d6dd46a1d9558c32ed17f1e9ce1726b272b06404 Mon Sep 17 00:00:00 2001 From: Alexander Polcyn Date: Mon, 10 Apr 2017 16:45:02 -0700 Subject: [PATCH 1/6] fix flakey race in ruby tests --- src/ruby/spec/generic/rpc_server_pool_spec.rb | 43 +++++++++++-------- 1 file changed, 26 insertions(+), 17 deletions(-) diff --git a/src/ruby/spec/generic/rpc_server_pool_spec.rb b/src/ruby/spec/generic/rpc_server_pool_spec.rb index 69e8222cb97..75b40225680 100644 --- a/src/ruby/spec/generic/rpc_server_pool_spec.rb +++ b/src/ruby/spec/generic/rpc_server_pool_spec.rb @@ -52,27 +52,36 @@ describe GRPC::Pool do expect(p.ready_for_work?).to be(false) end - it 'it stops being ready after all workers jobs waiting or running' do + it 'it stops being ready after all workers are busy, ' \ + 'and it becomes ready again after jobs complete' do p = Pool.new(5) p.start - job = proc { sleep(3) } # sleep so workers busy when done scheduling - 5.times do - expect(p.ready_for_work?).to be(true) - p.schedule(&job) - end - expect(p.ready_for_work?).to be(false) - end - it 'it becomes ready again after jobs complete' do - p = Pool.new(5) - p.start - job = proc {} + wait_mu = Mutex.new + wait_cv = ConditionVariable.new + wait = true + + job = proc do + wait_mu.synchronize do + wait_cv.wait(wait_mu) while wait + end + end + 5.times do expect(p.ready_for_work?).to be(true) p.schedule(&job) end + expect(p.ready_for_work?).to be(false) - sleep 5 # give the pool time do get at least one task done + + wait_mu.synchronize do + wait = false + wait_cv.broadcast + end + + # There's a potential race here but it shouldn't ever be + # reached with a 5 second sleep. + sleep 5 expect(p.ready_for_work?).to be(true) end end @@ -105,13 +114,13 @@ describe GRPC::Pool do it 'stops jobs when there are long running jobs' do p = Pool.new(1) p.start - o, q = Object.new, Queue.new + job_running = Queue.new job = proc do - sleep(5) # long running - q.push(o) + job_running.push(Object.new) + sleep(5000) end p.schedule(&job) - sleep(1) # should ensure the long job gets scheduled + job_running.pop expect { p.stop }.not_to raise_error end end From 9b020019497f5fb9fb035027d55d350450fb4ed0 Mon Sep 17 00:00:00 2001 From: Alexander Polcyn Date: Tue, 11 Apr 2017 10:14:13 -0700 Subject: [PATCH 2/6] get rid of racey sleep 5 and use a cv to wait forever --- src/ruby/spec/generic/rpc_server_pool_spec.rb | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/ruby/spec/generic/rpc_server_pool_spec.rb b/src/ruby/spec/generic/rpc_server_pool_spec.rb index 75b40225680..0803ca74ed6 100644 --- a/src/ruby/spec/generic/rpc_server_pool_spec.rb +++ b/src/ruby/spec/generic/rpc_server_pool_spec.rb @@ -52,8 +52,7 @@ describe GRPC::Pool do expect(p.ready_for_work?).to be(false) end - it 'it stops being ready after all workers are busy, ' \ - 'and it becomes ready again after jobs complete' do + it 'it stops being ready after all workers are busy' do p = Pool.new(5) p.start @@ -78,11 +77,6 @@ describe GRPC::Pool do wait = false wait_cv.broadcast end - - # There's a potential race here but it shouldn't ever be - # reached with a 5 second sleep. - sleep 5 - expect(p.ready_for_work?).to be(true) end end @@ -114,10 +108,17 @@ describe GRPC::Pool do it 'stops jobs when there are long running jobs' do p = Pool.new(1) p.start + + wait_forever_mu = Mutex.new + wait_forever_cv = ConditionVariable.new + wait_forever = true + job_running = Queue.new job = proc do job_running.push(Object.new) - sleep(5000) + wait_forever_mu.synchronize do + wait_forever_cv.wait while wait_forever + end end p.schedule(&job) job_running.pop From a03d204dd73621f87043b0f63ab646ba92169a8d Mon Sep 17 00:00:00 2001 From: David Garcia Quintas Date: Tue, 11 Apr 2017 16:09:31 -0700 Subject: [PATCH 3/6] Update credentials test after #9328 --- test/core/security/credentials_test.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/core/security/credentials_test.c b/test/core/security/credentials_test.c index c0933c64a68..9bdce71ba0c 100644 --- a/test/core/security/credentials_test.c +++ b/test/core/security/credentials_test.c @@ -582,7 +582,7 @@ static void on_oauth2_creds_get_metadata_failure( static void validate_compute_engine_http_request( const grpc_httpcli_request *request) { GPR_ASSERT(request->handshaker != &grpc_httpcli_ssl); - GPR_ASSERT(strcmp(request->host, "metadata") == 0); + GPR_ASSERT(strcmp(request->host, "metadata.google.internal") == 0); GPR_ASSERT( strcmp(request->http.path, "/computeMetadata/v1/instance/service-accounts/default/token") == From 75bfb9754827ef63ede77e27ad901e3355536419 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 11 Apr 2017 17:55:12 -0700 Subject: [PATCH 4/6] Finish hybrid server stuff, ensure it gets tested --- .../grpc++/impl/codegen/client_unary_call.h | 4 ++- .../grpc++/impl/codegen/completion_queue.h | 32 +++++++++---------- include/grpc++/impl/codegen/core_codegen.h | 9 +++++- .../impl/codegen/core_codegen_interface.h | 6 ++++ include/grpc++/impl/codegen/sync_stream.h | 12 +++++-- include/grpc/grpc.h | 23 ------------- include/grpc/impl/codegen/grpc_types.h | 23 +++++++++++++ src/core/lib/surface/completion_queue.c | 21 +++++++----- src/core/lib/surface/server.c | 7 ++-- src/cpp/common/core_codegen.cc | 12 +++++++ src/cpp/server/server_builder.cc | 26 +++++++++++---- test/cpp/end2end/async_end2end_test.cc | 28 ++++++++++++---- 12 files changed, 134 insertions(+), 69 deletions(-) diff --git a/include/grpc++/impl/codegen/client_unary_call.h b/include/grpc++/impl/codegen/client_unary_call.h index a5a4f3d7398..4bf35ae7785 100644 --- a/include/grpc++/impl/codegen/client_unary_call.h +++ b/include/grpc++/impl/codegen/client_unary_call.h @@ -52,7 +52,9 @@ template Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method, ClientContext* context, const InputMessage& request, OutputMessage* result) { - CompletionQueue cq(true); // Pluckable completion queue + CompletionQueue cq(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}); // Pluckable completion queue Call call(channel->CreateCall(method, context, &cq)); CallOpSet, diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h index 906a64693e1..c8ab726b0f4 100644 --- a/include/grpc++/impl/codegen/completion_queue.h +++ b/include/grpc++/impl/codegen/completion_queue.h @@ -102,7 +102,9 @@ class CompletionQueue : private GrpcLibraryCodegen { public: /// Default constructor. Implicitly creates a \a grpc_completion_queue /// instance. - CompletionQueue() : CompletionQueue(false) {} + CompletionQueue() + : CompletionQueue(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}) {} /// Wrap \a take, taking ownership of the instance. /// @@ -182,6 +184,16 @@ class CompletionQueue : private GrpcLibraryCodegen { }; void CompleteAvalanching(); + protected: + /// Private constructor of CompletionQueue only visible to friend classes + CompletionQueue(const grpc_completion_queue_attributes& attributes) { + cq_ = g_core_codegen_interface->grpc_completion_queue_create( + g_core_codegen_interface->grpc_completion_queue_factory_lookup( + &attributes), + &attributes, NULL); + InitialAvalanching(); // reserve this for the future shutdown + } + private: // Friend synchronous wrappers so that they can access Pluck(), which is // a semi-private API geared towards the synchronous implementation. @@ -215,18 +227,6 @@ class CompletionQueue : private GrpcLibraryCodegen { const InputMessage& request, OutputMessage* result); - /// Private constructor of CompletionQueue only visible to friend classes - CompletionQueue(bool is_pluck) { - if (is_pluck) { - cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_pluck( - nullptr); - } else { - cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_next( - nullptr); - } - InitialAvalanching(); // reserve this for the future shutdown - } - NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline); /// Wraps \a grpc_completion_queue_pluck. @@ -299,11 +299,9 @@ class ServerCompletionQueue : public CompletionQueue { /// AsyncNext()). By default all server completion queues are assumed to be /// frequently polled. ServerCompletionQueue(grpc_cq_polling_type polling_type) - : CompletionQueue(MakeCompletionQueue(polling_type)), + : CompletionQueue(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}), polling_type_(polling_type) {} - - static grpc_completion_queue* MakeCompletionQueue( - grpc_cq_polling_type polling_type); }; } // namespace grpc diff --git a/include/grpc++/impl/codegen/core_codegen.h b/include/grpc++/impl/codegen/core_codegen.h index 65151590b25..3cb7da8ef6c 100644 --- a/include/grpc++/impl/codegen/core_codegen.h +++ b/include/grpc++/impl/codegen/core_codegen.h @@ -44,8 +44,15 @@ namespace grpc { /// Implementation of the core codegen interface. -class CoreCodegen : public CoreCodegenInterface { +class CoreCodegen final : public CoreCodegenInterface { private: + virtual const grpc_completion_queue_factory* + grpc_completion_queue_factory_lookup( + const grpc_completion_queue_attributes* attributes) override; + virtual grpc_completion_queue* grpc_completion_queue_create( + const grpc_completion_queue_factory* factory, + const grpc_completion_queue_attributes* attributes, + void* reserved) override; grpc_completion_queue* grpc_completion_queue_create_for_next( void* reserved) override; grpc_completion_queue* grpc_completion_queue_create_for_pluck( diff --git a/include/grpc++/impl/codegen/core_codegen_interface.h b/include/grpc++/impl/codegen/core_codegen_interface.h index 529bef687bb..a1a0aaf3cad 100644 --- a/include/grpc++/impl/codegen/core_codegen_interface.h +++ b/include/grpc++/impl/codegen/core_codegen_interface.h @@ -59,6 +59,12 @@ class CoreCodegenInterface { virtual void assert_fail(const char* failed_assertion, const char* file, int line) = 0; + virtual const grpc_completion_queue_factory* + grpc_completion_queue_factory_lookup( + const grpc_completion_queue_attributes* attributes) = 0; + virtual grpc_completion_queue* grpc_completion_queue_create( + const grpc_completion_queue_factory* factory, + const grpc_completion_queue_attributes* attributes, void* reserved) = 0; virtual grpc_completion_queue* grpc_completion_queue_create_for_next( void* reserved) = 0; virtual grpc_completion_queue* grpc_completion_queue_create_for_pluck( diff --git a/include/grpc++/impl/codegen/sync_stream.h b/include/grpc++/impl/codegen/sync_stream.h index 328d5cb1e83..a010924cefe 100644 --- a/include/grpc++/impl/codegen/sync_stream.h +++ b/include/grpc++/impl/codegen/sync_stream.h @@ -156,7 +156,9 @@ class ClientReader final : public ClientReaderInterface { ClientReader(ChannelInterface* channel, const RpcMethod& method, ClientContext* context, const W& request) : context_(context), - cq_(true), // Pluckable cq + cq_(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { CallOpSet @@ -230,7 +232,9 @@ class ClientWriter : public ClientWriterInterface { ClientWriter(ChannelInterface* channel, const RpcMethod& method, ClientContext* context, R* response) : context_(context), - cq_(true), // Pluckable cq + cq_(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { finish_ops_.RecvMessage(response); finish_ops_.AllowNoMessage(); @@ -330,7 +334,9 @@ class ClientReaderWriter final : public ClientReaderWriterInterface { ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method, ClientContext* context) : context_(context), - cq_(true), // Pluckable cq + cq_(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { if (!context_->initial_metadata_corked_) { CallOpSet ops; diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index f3201edad29..1a7d0120bfa 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -93,29 +93,6 @@ GRPCAPI const char *grpc_version_string(void); /** Return a string specifying what the 'g' in gRPC stands for */ GRPCAPI const char *grpc_g_stands_for(void); -/** Specifies the type of APIs to use to pop events from the completion queue */ -typedef enum { - /** Events are popped out by calling grpc_completion_queue_next() API ONLY */ - GRPC_CQ_NEXT = 1, - - /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/ - GRPC_CQ_PLUCK -} grpc_cq_completion_type; - -#define GRPC_CQ_CURRENT_VERSION 1 -typedef struct grpc_completion_queue_attributes { - /* The version number of this structure. More fields might be added to this - structure in future. */ - int version; /* Set to GRPC_CQ_CURRENT_VERSION */ - - grpc_cq_completion_type cq_completion_type; - - grpc_cq_polling_type cq_polling_type; -} grpc_completion_queue_attributes; - -/** The completion queue factory structure is opaque to the callers of grpc */ -typedef struct grpc_completion_queue_factory grpc_completion_queue_factory; - /** Returns the completion queue factory based on the attributes. MAY return a NULL if no factory can be found */ GRPCAPI const grpc_completion_queue_factory * diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index 02732205f5f..a8eda739bf6 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -569,6 +569,29 @@ typedef enum { GRPC_CQ_NON_POLLING } grpc_cq_polling_type; +/** Specifies the type of APIs to use to pop events from the completion queue */ +typedef enum { + /** Events are popped out by calling grpc_completion_queue_next() API ONLY */ + GRPC_CQ_NEXT = 1, + + /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/ + GRPC_CQ_PLUCK +} grpc_cq_completion_type; + +#define GRPC_CQ_CURRENT_VERSION 1 +typedef struct grpc_completion_queue_attributes { + /* The version number of this structure. More fields might be added to this + structure in future. */ + int version; /* Set to GRPC_CQ_CURRENT_VERSION */ + + grpc_cq_completion_type cq_completion_type; + + grpc_cq_polling_type cq_polling_type; +} grpc_completion_queue_attributes; + +/** The completion queue factory structure is opaque to the callers of grpc */ +typedef struct grpc_completion_queue_factory grpc_completion_queue_factory; + #ifdef __cplusplus } #endif diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index e17f094837e..ea97a6f374b 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -61,6 +61,7 @@ typedef struct { } plucker; typedef struct { + bool can_get_pollset; size_t (*size)(void); void (*init)(grpc_pollset *pollset, gpr_mu **mu); grpc_error *(*kick)(grpc_pollset *pollset, @@ -107,9 +108,10 @@ static grpc_error *non_polling_poller_work(grpc_exec_ctx *exec_ctx, 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); - *worker = (grpc_pollset_worker *)&w; + if (worker != NULL) *worker = (grpc_pollset_worker *)&w; if (npp->root == NULL) { npp->root = w.next = w.prev = &w; } else { @@ -128,11 +130,11 @@ static grpc_error *non_polling_poller_work(grpc_exec_ctx *exec_ctx, } npp->root = NULL; } - w.next->prev = w.prev; - w.prev->next = w.next; } + w.next->prev = w.prev; + w.prev->next = w.next; gpr_cv_destroy(&w.cv); - *worker = NULL; + if (worker != NULL) *worker = NULL; return GRPC_ERROR_NONE; } @@ -169,21 +171,24 @@ static void non_polling_poller_shutdown(grpc_exec_ctx *exec_ctx, static const cq_poller_vtable g_poller_vtable_by_poller_type[] = { /* GRPC_CQ_DEFAULT_POLLING */ - {.size = grpc_pollset_size, + {.can_get_pollset = 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 */ - {.size = grpc_pollset_size, + {.can_get_pollset = 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_POLLING */ - {.size = non_polling_poller_size, + {.can_get_pollset = false, + .size = non_polling_poller_size, .init = non_polling_poller_init, .kick = non_polling_poller_kick, .work = non_polling_poller_work, @@ -837,7 +842,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) { diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 9496f90390d..767c91a5ec5 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -1009,6 +1009,8 @@ void grpc_server_register_completion_queue(grpc_server *server, calls grpc_completion_queue_pluck() on server completion queues */ } + GPR_ASSERT(grpc_cq_pollset(cq)); + register_completion_queue(server, cq, false, reserved); } @@ -1102,8 +1104,9 @@ void grpc_server_start(grpc_server *server) { 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])) { - server->pollsets[server->pollset_count++] = - grpc_cq_pollset(server->cqs[i]); + grpc_pollset *pollset = grpc_cq_pollset(server->cqs[i]); + GPR_ASSERT(pollset); + server->pollsets[server->pollset_count++] = pollset; } server->request_freelist_per_cq[i] = gpr_stack_lockfree_create((size_t)server->max_requested_calls_per_cq); diff --git a/src/cpp/common/core_codegen.cc b/src/cpp/common/core_codegen.cc index 0dd758ec4e5..8f1de222fb8 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 c6784ea1593..6687fe78b92 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -243,6 +243,16 @@ std::unique_ptr ServerBuilder::BuildAndStart() { sync_server_cqs(std::make_shared< std::vector>>()); + 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, @@ -253,7 +263,7 @@ std::unique_ptr ServerBuilder::BuildAndStart() { sync_server_settings_.cq_timeout_msec); grpc_cq_polling_type polling_type = - cqs_.empty() ? GRPC_CQ_DEFAULT_POLLING : GRPC_CQ_NON_POLLING; + 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++) { @@ -273,12 +283,15 @@ std::unique_ptr 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); + if (is_hybrid_server) { + grpc_server_register_non_listening_completion_queue(server->server_, + (*it)->cq(), nullptr); + } else { + 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 @@ -290,7 +303,6 @@ std::unique_ptr ServerBuilder::BuildAndStart() { 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); diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 0b5215ef8e4..cc3958bf13c 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -38,6 +38,7 @@ #include #include #include +#include #include #include #include @@ -49,6 +50,7 @@ #include #include "src/core/lib/iomgr/port.h" +#include "src/proto/grpc/health/v1/health.grpc.pb.h" #include "src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.h" #include "src/proto/grpc/testing/echo.grpc.pb.h" #include "test/core/util/port.h" @@ -224,13 +226,15 @@ class ServerBuilderSyncPluginDisabler : public ::grpc::ServerBuilderOption { class TestScenario { public: - TestScenario(bool non_block, const grpc::string& creds_type, + TestScenario(bool non_block, const grpc::string& creds_type, bool hcs, const grpc::string& content) : disable_blocking(non_block), + health_check_service(hcs), credentials_type(creds_type), message_content(content) {} void Log() const; bool disable_blocking; + bool health_check_service; // Although the below grpc::string's are logically const, we can't declare // them const because of a limitation in the way old compilers (e.g., gcc-4.4) // manage vector insertion using a copy constructor @@ -243,6 +247,8 @@ static std::ostream& operator<<(std::ostream& out, return out << "TestScenario{disable_blocking=" << (scenario.disable_blocking ? "true" : "false") << ", credentials='" << scenario.credentials_type + << ", health_check_service=" + << (scenario.health_check_service ? "true" : "false") << "', message_size=" << scenario.message_content.size() << "}"; } @@ -252,6 +258,8 @@ void TestScenario::Log() const { gpr_log(GPR_DEBUG, "%s", out.str().c_str()); } +class HealthCheck : public health::v1::Health::Service {}; + class AsyncEnd2endTest : public ::testing::TestWithParam { protected: AsyncEnd2endTest() { GetParam().Log(); } @@ -268,6 +276,9 @@ class AsyncEnd2endTest : public ::testing::TestWithParam { GetParam().credentials_type); builder.AddListeningPort(server_address_.str(), server_creds); builder.RegisterService(&service_); + if (GetParam().health_check_service) { + builder.RegisterService(&health_check_); + } cq_ = builder.AddCompletionQueue(); // TODO(zyc): make a test option to choose wheather sync plugins should be @@ -340,6 +351,7 @@ class AsyncEnd2endTest : public ::testing::TestWithParam { std::unique_ptr stub_; std::unique_ptr server_; grpc::testing::EchoTestService::AsyncService service_; + HealthCheck health_check_; std::ostringstream server_address_; int port_; @@ -1754,12 +1766,14 @@ std::vector CreateTestScenarios(bool test_disable_blocking, messages.push_back(big_msg); } - for (auto cred = credentials_types.begin(); cred != credentials_types.end(); - ++cred) { - for (auto msg = messages.begin(); msg != messages.end(); msg++) { - scenarios.emplace_back(false, *cred, *msg); - if (test_disable_blocking) { - scenarios.emplace_back(true, *cred, *msg); + for (auto health_check_service : {false, true}) { + for (auto cred = credentials_types.begin(); cred != credentials_types.end(); + ++cred) { + for (auto msg = messages.begin(); msg != messages.end(); msg++) { + scenarios.emplace_back(false, *cred, health_check_service, *msg); + if (test_disable_blocking) { + scenarios.emplace_back(true, *cred, health_check_service, *msg); + } } } } From 58aa706aaf1c39e092f246202b18e6a2931dc664 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 12 Apr 2017 08:10:58 -0700 Subject: [PATCH 5/6] Fix registration --- src/core/lib/surface/server.c | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 767c91a5ec5..1680085f67b 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -1009,8 +1009,6 @@ void grpc_server_register_completion_queue(grpc_server *server, calls grpc_completion_queue_pluck() on server completion queues */ } - GPR_ASSERT(grpc_cq_pollset(cq)); - register_completion_queue(server, cq, false, reserved); } @@ -1105,8 +1103,7 @@ void grpc_server_start(grpc_server *server) { for (i = 0; i < server->cq_count; i++) { if (!grpc_cq_is_non_listening_server_cq(server->cqs[i])) { grpc_pollset *pollset = grpc_cq_pollset(server->cqs[i]); - GPR_ASSERT(pollset); - server->pollsets[server->pollset_count++] = pollset; + if (pollset != NULL) server->pollsets[server->pollset_count++] = pollset; } server->request_freelist_per_cq[i] = gpr_stack_lockfree_create((size_t)server->max_requested_calls_per_cq); From 11c5832b3e8be35f16465d8ef38a1d0ba033a822 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 12 Apr 2017 08:21:17 -0700 Subject: [PATCH 6/6] Get rid of second api for marking non-listening cqs --- include/grpc/grpc.h | 9 --------- src/core/lib/surface/completion_queue.c | 12 +++++++++++- src/core/lib/surface/completion_queue.h | 5 ++--- src/core/lib/surface/server.c | 23 +++++------------------ src/cpp/server/server_builder.cc | 18 ++++-------------- 5 files changed, 22 insertions(+), 45 deletions(-) diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index 1a7d0120bfa..7b37f34acc4 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -376,15 +376,6 @@ GRPCAPI void grpc_server_register_completion_queue(grpc_server *server, grpc_completion_queue *cq, void *reserved); -/** Register a non-listening completion queue with the server. This API is - similar to grpc_server_register_completion_queue except that the server will - not use this completion_queue to listen to any incoming channels. - - Registering a non-listening completion queue will have negative performance - impact and hence this API is not recommended for production use cases. */ -GRPCAPI void grpc_server_register_non_listening_completion_queue( - grpc_server *server, grpc_completion_queue *q, void *reserved); - /** Add a HTTP2 over plaintext over tcp listener. Returns bound port number on success, 0 on failure. REQUIRES: server not started */ diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index ea97a6f374b..eae3f103b12 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -62,6 +62,7 @@ typedef struct { 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, @@ -172,6 +173,7 @@ static void non_polling_poller_shutdown(grpc_exec_ctx *exec_ctx, 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, @@ -180,6 +182,7 @@ static const cq_poller_vtable g_poller_vtable_by_poller_type[] = { .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, @@ -188,6 +191,7 @@ static const cq_poller_vtable g_poller_vtable_by_poller_type[] = { .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, @@ -863,4 +867,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 0995a56889b..a932087939d 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -94,10 +94,9 @@ 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); diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 1680085f67b..da8b6339b2b 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -974,7 +974,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++) { @@ -983,10 +983,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, @@ -1009,16 +1005,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) { @@ -1101,9 +1088,9 @@ 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])) { - grpc_pollset *pollset = grpc_cq_pollset(server->cqs[i]); - if (pollset != NULL) server->pollsets[server->pollset_count++] = pollset; + if (grpc_cq_can_listen(server->cqs[i])) { + server->pollsets[server->pollset_count++] = + grpc_cq_pollset(server->cqs[i]); } server->request_freelist_per_cq[i] = gpr_stack_lockfree_create((size_t)server->max_requested_calls_per_cq); diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index 6687fe78b92..a92cec643ce 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -284,13 +284,8 @@ std::unique_ptr ServerBuilder::BuildAndStart() { // 2. cqs_: Completion queues added via AddCompletionQueue() call for (auto it = sync_server_cqs->begin(); it != sync_server_cqs->end(); ++it) { - if (is_hybrid_server) { - grpc_server_register_non_listening_completion_queue(server->server_, - (*it)->cq(), nullptr); - } else { - grpc_server_register_completion_queue(server->server_, (*it)->cq(), - nullptr); - } + grpc_server_register_completion_queue(server->server_, (*it)->cq(), + nullptr); num_frequently_polled_cqs++; } @@ -300,13 +295,8 @@ std::unique_ptr 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); - } 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) {