From 2f40ff423ece50964d1041a4ad68939260da9fe6 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 11 Apr 2017 16:01:19 -0700 Subject: Support making hybrid cqs in core --- src/core/lib/surface/completion_queue.c | 167 ++++++++++++++++++++++++++++---- 1 file changed, 149 insertions(+), 18 deletions(-) (limited to 'src/core/lib/surface/completion_queue.c') diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index 35e9f7eb30..e17f094837 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -60,13 +60,145 @@ typedef struct { void *tag; } plucker; +typedef struct { + 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; + non_polling_worker w; + gpr_cv_init(&w.cv); + *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); + *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 */ + {.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, + .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, + .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}, +}; + /* Completion queue structure */ struct grpc_completion_queue { /** owned by pollset */ gpr_mu *mu; grpc_cq_completion_type completion_type; - grpc_cq_polling_type polling_type; + + const cq_poller_vtable *poller_vtable; /** completed events */ grpc_cq_completion completed_head; @@ -127,15 +259,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); @@ -164,10 +299,6 @@ grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc) { return cc->completion_type; } -grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc) { - return cc->polling_type; -} - #ifdef GRPC_CQ_REF_COUNT_DEBUG void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason, const char *file, int line) { @@ -195,7 +326,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)); #ifndef NDEBUG gpr_free(cc->outstanding_tags); #endif @@ -280,7 +411,7 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, } } 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) { const char *msg = grpc_error_string(kick_error); @@ -295,8 +426,8 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, GPR_ASSERT(!cc->shutdown); GPR_ASSERT(cc->shutdown_called); 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); } @@ -452,8 +583,8 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, gpr_mu_lock(cc->mu); continue; } else { - 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); if (err != GRPC_ERROR_NONE) { gpr_mu_unlock(cc->mu); const char *msg = grpc_error_string(err); @@ -644,8 +775,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); @@ -689,8 +820,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); -- cgit v1.2.3 From 75bfb9754827ef63ede77e27ad901e3355536419 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 11 Apr 2017 17:55:12 -0700 Subject: Finish hybrid server stuff, ensure it gets tested --- include/grpc++/impl/codegen/client_unary_call.h | 4 ++- include/grpc++/impl/codegen/completion_queue.h | 32 ++++++++++------------ include/grpc++/impl/codegen/core_codegen.h | 9 +++++- .../grpc++/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(-) (limited to 'src/core/lib/surface/completion_queue.c') diff --git a/include/grpc++/impl/codegen/client_unary_call.h b/include/grpc++/impl/codegen/client_unary_call.h index a5a4f3d739..4bf35ae778 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 906a64693e..c8ab726b0f 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 65151590b2..3cb7da8ef6 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 529bef687b..a1a0aaf3ca 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 328d5cb1e8..a010924cef 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 f3201edad2..1a7d0120bf 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 02732205f5..a8eda739bf 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 e17f094837..ea97a6f374 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 9496f90390..767c91a5ec 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 0dd758ec4e..8f1de222fb 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 c6784ea159..6687fe78b9 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 0b5215ef8e..cc3958bf13 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); + } } } } -- cgit v1.2.3 From 11c5832b3e8be35f16465d8ef38a1d0ba033a822 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 12 Apr 2017 08:21:17 -0700 Subject: 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(-) (limited to 'src/core/lib/surface/completion_queue.c') diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index 1a7d0120bf..7b37f34acc 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 ea97a6f374..eae3f103b1 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 0995a56889..a932087939 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 1680085f67..da8b6339b2 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 6687fe78b9..a92cec643c 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) { -- cgit v1.2.3