From bdd13cb0aef7d3f6dbc467148b4b3158485359eb Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Wed, 1 Aug 2018 11:22:40 -0700 Subject: Revert "Revert "Restrict the number of threads in C++ sync server"" --- test/core/surface/public_headers_must_be_c89.c | 1 + 1 file changed, 1 insertion(+) (limited to 'test/core/surface') diff --git a/test/core/surface/public_headers_must_be_c89.c b/test/core/surface/public_headers_must_be_c89.c index 9f4ad2b4d7..497f7194d5 100644 --- a/test/core/surface/public_headers_must_be_c89.c +++ b/test/core/surface/public_headers_must_be_c89.c @@ -130,6 +130,7 @@ int main(int argc, char **argv) { printf("%lx", (unsigned long) grpc_resource_quota_ref); printf("%lx", (unsigned long) grpc_resource_quota_unref); printf("%lx", (unsigned long) grpc_resource_quota_resize); + printf("%lx", (unsigned long) grpc_resource_quota_set_max_threads); printf("%lx", (unsigned long) grpc_resource_quota_arg_vtable); printf("%lx", (unsigned long) grpc_channelz_get_top_channels); printf("%lx", (unsigned long) grpc_channelz_get_channel); -- cgit v1.2.3 From e84096bbe5f0e471d90906e93cba9332c392aa60 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Thu, 9 Aug 2018 13:20:35 -0700 Subject: Experimental infrastructure for callback-based CQ --- grpc.def | 1 + include/grpc/grpc.h | 6 + include/grpc/impl/codegen/grpc_types.h | 19 ++- include/grpcpp/impl/codegen/client_unary_call.h | 4 +- include/grpcpp/impl/codegen/completion_queue.h | 5 +- include/grpcpp/impl/codegen/sync_stream.h | 12 +- src/core/lib/surface/completion_queue.cc | 150 +++++++++++++++++++-- src/core/lib/surface/completion_queue.h | 13 +- src/core/lib/surface/completion_queue_factory.cc | 17 ++- .../GRPCClient/private/GRPCCompletionQueue.m | 6 +- src/ruby/ext/grpc/rb_grpc_imports.generated.c | 2 + src/ruby/ext/grpc/rb_grpc_imports.generated.h | 3 + test/core/surface/public_headers_must_be_c89.c | 1 + 13 files changed, 211 insertions(+), 28 deletions(-) (limited to 'test/core/surface') diff --git a/grpc.def b/grpc.def index 5e9d86c769..962a2ec716 100644 --- a/grpc.def +++ b/grpc.def @@ -20,6 +20,7 @@ EXPORTS grpc_completion_queue_factory_lookup grpc_completion_queue_create_for_next grpc_completion_queue_create_for_pluck + grpc_completion_queue_create_for_callback grpc_completion_queue_create grpc_completion_queue_next grpc_completion_queue_pluck diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index 348c7a316f..4c3af45100 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -101,6 +101,12 @@ GRPCAPI grpc_completion_queue* grpc_completion_queue_create_for_next( GRPCAPI grpc_completion_queue* grpc_completion_queue_create_for_pluck( void* reserved); +/** Helper function to create a completion queue with grpc_cq_completion_type + of GRPC_CQ_CALLBACK and grpc_cq_polling_type of GRPC_CQ_DEFAULT_POLLING. + This function is experimental. */ +GRPCAPI grpc_completion_queue* grpc_completion_queue_create_for_callback( + void* shutdown_callback, void* reserved); + /** Create a completion queue */ GRPCAPI grpc_completion_queue* grpc_completion_queue_create( const grpc_completion_queue_factory* factory, diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index 5fd080c48b..b5353c1dea 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -651,10 +651,16 @@ typedef enum { GRPC_CQ_NEXT, /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/ - GRPC_CQ_PLUCK + GRPC_CQ_PLUCK, + + /** EXPERIMENTAL: Events trigger a callback specified as the tag */ + GRPC_CQ_CALLBACK } grpc_cq_completion_type; -#define GRPC_CQ_CURRENT_VERSION 1 +/* The upgrade to version 2 is currently experimental. */ + +#define GRPC_CQ_CURRENT_VERSION 2 +#define GRPC_CQ_VERSION_MINIMUM_FOR_CALLBACKABLE 2 typedef struct grpc_completion_queue_attributes { /** The version number of this structure. More fields might be added to this structure in future. */ @@ -663,6 +669,15 @@ typedef struct grpc_completion_queue_attributes { grpc_cq_completion_type cq_completion_type; grpc_cq_polling_type cq_polling_type; + + /* END OF VERSION 1 CQ ATTRIBUTES */ + + /* EXPERIMENTAL: START OF VERSION 2 CQ ATTRIBUTES */ + /** When creating a callbackable CQ, pass in a functor to get invoked when + * shutdown is complete */ + void* cq_shutdown_cb; + + /* END OF VERSION 2 CQ ATTRIBUTES */ } grpc_completion_queue_attributes; /** The completion queue factory structure is opaque to the callers of grpc */ diff --git a/include/grpcpp/impl/codegen/client_unary_call.h b/include/grpcpp/impl/codegen/client_unary_call.h index a37a81b75b..e4e8364e07 100644 --- a/include/grpcpp/impl/codegen/client_unary_call.h +++ b/include/grpcpp/impl/codegen/client_unary_call.h @@ -50,8 +50,8 @@ class BlockingUnaryCallImpl { ClientContext* context, const InputMessage& request, OutputMessage* result) { CompletionQueue cq(grpc_completion_queue_attributes{ - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, - GRPC_CQ_DEFAULT_POLLING}); // Pluckable completion queue + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING, + nullptr}); // Pluckable completion queue Call call(channel->CreateCall(method, context, &cq)); CallOpSet, diff --git a/include/grpcpp/impl/codegen/completion_queue.h b/include/grpcpp/impl/codegen/completion_queue.h index 5819e068ba..272575dac2 100644 --- a/include/grpcpp/impl/codegen/completion_queue.h +++ b/include/grpcpp/impl/codegen/completion_queue.h @@ -97,7 +97,8 @@ class CompletionQueue : private GrpcLibraryCodegen { /// instance. CompletionQueue() : CompletionQueue(grpc_completion_queue_attributes{ - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}) {} + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING, + nullptr}) {} /// Wrap \a take, taking ownership of the instance. /// @@ -376,7 +377,7 @@ class ServerCompletionQueue : public CompletionQueue { /// frequently polled. ServerCompletionQueue(grpc_cq_polling_type polling_type) : CompletionQueue(grpc_completion_queue_attributes{ - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}), + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type, nullptr}), polling_type_(polling_type) {} grpc_cq_polling_type polling_type_; diff --git a/include/grpcpp/impl/codegen/sync_stream.h b/include/grpcpp/impl/codegen/sync_stream.h index 7152eaf41f..cbfcf25d0a 100644 --- a/include/grpcpp/impl/codegen/sync_stream.h +++ b/include/grpcpp/impl/codegen/sync_stream.h @@ -243,8 +243,8 @@ class ClientReader final : public ClientReaderInterface { ClientContext* context, const W& request) : context_(context), cq_(grpc_completion_queue_attributes{ - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, - GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING, + nullptr}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, ::grpc::internal::CallOpSendMessage, @@ -377,8 +377,8 @@ class ClientWriter : public ClientWriterInterface { ClientContext* context, R* response) : context_(context), cq_(grpc_completion_queue_attributes{ - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, - GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING, + nullptr}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { finish_ops_.RecvMessage(response); finish_ops_.AllowNoMessage(); @@ -551,8 +551,8 @@ class ClientReaderWriter final : public ClientReaderWriterInterface { ClientContext* context) : context_(context), cq_(grpc_completion_queue_attributes{ - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, - GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING, + nullptr}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { if (!context_->initial_metadata_corked_) { ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> diff --git a/src/core/lib/surface/completion_queue.cc b/src/core/lib/surface/completion_queue.cc index 7da9e6b74c..fd33ce044c 100644 --- a/src/core/lib/surface/completion_queue.cc +++ b/src/core/lib/surface/completion_queue.cc @@ -184,7 +184,7 @@ static const cq_poller_vtable g_poller_vtable_by_poller_type[] = { typedef struct cq_vtable { grpc_cq_completion_type cq_completion_type; size_t data_size; - void (*init)(void* data); + void (*init)(void* data, grpc_core::CQCallbackInterface* shutdown_callback); void (*shutdown)(grpc_completion_queue* cq); void (*destroy)(void* data); bool (*begin_op)(grpc_completion_queue* cq, void* tag); @@ -253,6 +253,29 @@ typedef struct cq_pluck_data { plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS]; } cq_pluck_data; +typedef struct cq_callback_data { + /** No actual completed events queue, unlike other types */ + + /** Number of pending events (+1 if we're not shutdown) */ + gpr_atm pending_events; + + /** Counter of how many things have ever been queued on this completion queue + useful for avoiding locks to check the queue */ + gpr_atm things_queued_ever; + + /** 0 initially. 1 once we completed shutting */ + /* TODO: (sreek) This is not needed since (shutdown == 1) if and only if + * (pending_events == 0). So consider removing this in future and use + * pending_events */ + gpr_atm shutdown; + + /** 0 initially. 1 once we initiated shutdown */ + bool shutdown_called; + + /** A callback that gets invoked when the CQ completes shutdown */ + grpc_core::CQCallbackInterface* shutdown_callback; +} cq_callback_data; + /* Completion queue structure */ struct grpc_completion_queue { /** Once owning_refs drops to zero, we will destroy the cq */ @@ -276,11 +299,14 @@ struct grpc_completion_queue { /* Forward declarations */ static void cq_finish_shutdown_next(grpc_completion_queue* cq); static void cq_finish_shutdown_pluck(grpc_completion_queue* cq); +static void cq_finish_shutdown_callback(grpc_completion_queue* cq); static void cq_shutdown_next(grpc_completion_queue* cq); static void cq_shutdown_pluck(grpc_completion_queue* cq); +static void cq_shutdown_callback(grpc_completion_queue* cq); static bool cq_begin_op_for_next(grpc_completion_queue* cq, void* tag); static bool cq_begin_op_for_pluck(grpc_completion_queue* cq, void* tag); +static bool cq_begin_op_for_callback(grpc_completion_queue* cq, void* tag); static void cq_end_op_for_next(grpc_completion_queue* cq, void* tag, grpc_error* error, @@ -294,16 +320,25 @@ static void cq_end_op_for_pluck(grpc_completion_queue* cq, void* tag, grpc_cq_completion* storage), void* done_arg, grpc_cq_completion* storage); +static void cq_end_op_for_callback(grpc_completion_queue* cq, void* tag, + grpc_error* error, + void (*done)(void* done_arg, + grpc_cq_completion* storage), + void* done_arg, grpc_cq_completion* storage); + static grpc_event cq_next(grpc_completion_queue* cq, gpr_timespec deadline, void* reserved); static grpc_event cq_pluck(grpc_completion_queue* cq, void* tag, gpr_timespec deadline, void* reserved); -static void cq_init_next(void* data); -static void cq_init_pluck(void* data); +static void cq_init_next(void* data, grpc_core::CQCallbackInterface*); +static void cq_init_pluck(void* data, grpc_core::CQCallbackInterface*); +static void cq_init_callback(void* data, + grpc_core::CQCallbackInterface* shutdown_callback); static void cq_destroy_next(void* data); static void cq_destroy_pluck(void* data); +static void cq_destroy_callback(void* data); /* Completion queue vtables based on the completion-type */ static const cq_vtable g_cq_vtable[] = { @@ -315,6 +350,10 @@ static const cq_vtable g_cq_vtable[] = { {GRPC_CQ_PLUCK, sizeof(cq_pluck_data), cq_init_pluck, cq_shutdown_pluck, cq_destroy_pluck, cq_begin_op_for_pluck, cq_end_op_for_pluck, nullptr, cq_pluck}, + /* GRPC_CQ_CALLBACK */ + {GRPC_CQ_CALLBACK, sizeof(cq_callback_data), cq_init_callback, + cq_shutdown_callback, cq_destroy_callback, cq_begin_op_for_callback, + cq_end_op_for_callback, nullptr, nullptr}, }; #define DATA_FROM_CQ(cq) ((void*)(cq + 1)) @@ -419,8 +458,8 @@ static long cq_event_queue_num_items(grpc_cq_event_queue* q) { } grpc_completion_queue* grpc_completion_queue_create_internal( - grpc_cq_completion_type completion_type, - grpc_cq_polling_type polling_type) { + grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type, + grpc_core::CQCallbackInterface* shutdown_callback) { GPR_TIMER_SCOPE("grpc_completion_queue_create_internal", 0); grpc_completion_queue* cq; @@ -448,14 +487,14 @@ grpc_completion_queue* grpc_completion_queue_create_internal( gpr_ref_init(&cq->owning_refs, 2); poller_vtable->init(POLLSET_FROM_CQ(cq), &cq->mu); - vtable->init(DATA_FROM_CQ(cq)); + vtable->init(DATA_FROM_CQ(cq), shutdown_callback); GRPC_CLOSURE_INIT(&cq->pollset_shutdown_done, on_pollset_shutdown_done, cq, grpc_schedule_on_exec_ctx); return cq; } -static void cq_init_next(void* ptr) { +static void cq_init_next(void* ptr, grpc_core::CQCallbackInterface*) { cq_next_data* cqd = static_cast(ptr); /* Initial count is dropped by grpc_completion_queue_shutdown */ gpr_atm_no_barrier_store(&cqd->pending_events, 1); @@ -470,7 +509,7 @@ static void cq_destroy_next(void* ptr) { cq_event_queue_destroy(&cqd->queue); } -static void cq_init_pluck(void* ptr) { +static void cq_init_pluck(void* ptr, grpc_core::CQCallbackInterface*) { cq_pluck_data* cqd = static_cast(ptr); /* Initial count is dropped by grpc_completion_queue_shutdown */ gpr_atm_no_barrier_store(&cqd->pending_events, 1); @@ -487,6 +526,19 @@ static void cq_destroy_pluck(void* ptr) { GPR_ASSERT(cqd->completed_head.next == (uintptr_t)&cqd->completed_head); } +static void cq_init_callback( + void* ptr, grpc_core::CQCallbackInterface* shutdown_callback) { + cq_callback_data* cqd = static_cast(ptr); + /* Initial count is dropped by grpc_completion_queue_shutdown */ + gpr_atm_no_barrier_store(&cqd->pending_events, 1); + gpr_atm_no_barrier_store(&cqd->shutdown, 0); + cqd->shutdown_called = false; + gpr_atm_no_barrier_store(&cqd->things_queued_ever, 0); + cqd->shutdown_callback = shutdown_callback; +} + +static void cq_destroy_callback(void* ptr) {} + grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue* cq) { return cq->vtable->cq_completion_type; } @@ -596,6 +648,11 @@ static bool cq_begin_op_for_pluck(grpc_completion_queue* cq, void* tag) { return atm_inc_if_nonzero(&cqd->pending_events); } +static bool cq_begin_op_for_callback(grpc_completion_queue* cq, void* tag) { + cq_callback_data* cqd = static_cast DATA_FROM_CQ(cq); + return atm_inc_if_nonzero(&cqd->pending_events); +} + bool grpc_cq_begin_op(grpc_completion_queue* cq, void* tag) { #ifndef NDEBUG gpr_mu_lock(cq->mu); @@ -759,6 +816,47 @@ static void cq_end_op_for_pluck(grpc_completion_queue* cq, void* tag, GRPC_ERROR_UNREF(error); } +/* Complete an event on a completion queue of type GRPC_CQ_CALLBACK */ +static void cq_end_op_for_callback( + grpc_completion_queue* cq, void* tag, grpc_error* error, + void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg, + grpc_cq_completion* storage) { + GPR_TIMER_SCOPE("cq_end_op_for_callback", 0); + + cq_callback_data* cqd = static_cast DATA_FROM_CQ(cq); + bool is_success = (error == GRPC_ERROR_NONE); + + if (grpc_api_trace.enabled() || + (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE)) { + const char* errmsg = grpc_error_string(error); + GRPC_API_TRACE( + "cq_end_op_for_callback(cq=%p, tag=%p, error=%s, " + "done=%p, done_arg=%p, storage=%p)", + 6, (cq, tag, errmsg, done, done_arg, storage)); + if (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE) { + gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag, errmsg); + } + } + + /* We don't care for the storage content */ + done(done_arg, storage); + + gpr_mu_lock(cq->mu); + cq_check_tag(cq, tag, false); /* Used in debug builds only */ + + gpr_atm_no_barrier_fetch_add(&cqd->things_queued_ever, 1); + if (gpr_atm_full_fetch_add(&cqd->pending_events, -1) == 1) { + cq_finish_shutdown_callback(cq); + gpr_mu_unlock(cq->mu); + } else { + gpr_mu_unlock(cq->mu); + } + + GRPC_ERROR_UNREF(error); + + (static_cast(tag))->Run(is_success); +} + void grpc_cq_end_op(grpc_completion_queue* cq, void* tag, grpc_error* error, void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg, grpc_cq_completion* storage) { @@ -1233,6 +1331,42 @@ static void cq_shutdown_pluck(grpc_completion_queue* cq) { GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (pluck cq)"); } +static void cq_finish_shutdown_callback(grpc_completion_queue* cq) { + cq_callback_data* cqd = static_cast DATA_FROM_CQ(cq); + auto* callback = cqd->shutdown_callback; + + GPR_ASSERT(cqd->shutdown_called); + GPR_ASSERT(!gpr_atm_no_barrier_load(&cqd->shutdown)); + gpr_atm_no_barrier_store(&cqd->shutdown, 1); + + cq->poller_vtable->shutdown(POLLSET_FROM_CQ(cq), &cq->pollset_shutdown_done); + callback->Run(true); +} + +static void cq_shutdown_callback(grpc_completion_queue* cq) { + cq_callback_data* cqd = static_cast DATA_FROM_CQ(cq); + + /* Need an extra ref for cq here because: + * We call cq_finish_shutdown_pluck() below, that would call pollset shutdown. + * Pollset shutdown decrements the cq ref count which can potentially destroy + * the cq (if that happens to be the last ref). + * Creating an extra ref here prevents the cq from getting destroyed while + * this function is still active */ + GRPC_CQ_INTERNAL_REF(cq, "shutting_down (callback cq)"); + gpr_mu_lock(cq->mu); + if (cqd->shutdown_called) { + gpr_mu_unlock(cq->mu); + GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (callback cq)"); + return; + } + cqd->shutdown_called = true; + if (gpr_atm_full_fetch_add(&cqd->pending_events, -1) == 1) { + cq_finish_shutdown_callback(cq); + } + gpr_mu_unlock(cq->mu); + GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (callback cq)"); +} + /* Shutdown simply drops a ref that we reserved at creation time; if we drop to zero here, then enter shutdown mode and wake up any waiters */ void grpc_completion_queue_shutdown(grpc_completion_queue* cq) { diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 84446a4d92..6d8c6c9b06 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -47,6 +47,16 @@ typedef struct grpc_cq_completion { uintptr_t next; } grpc_cq_completion; +/// For callback CQs, the following is what is actually intended by +/// the tag. +namespace grpc_core { +class CQCallbackInterface { + public: + virtual ~CQCallbackInterface() {} + virtual void Run(bool) = 0; +}; +} // namespace grpc_core + #ifndef NDEBUG void grpc_cq_internal_ref(grpc_completion_queue* cc, const char* reason, const char* file, int line); @@ -87,6 +97,7 @@ grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue* cc); int grpc_get_cq_poll_num(grpc_completion_queue* cc); grpc_completion_queue* grpc_completion_queue_create_internal( - grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type); + grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type, + grpc_core::CQCallbackInterface* shutdown_callback); #endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */ diff --git a/src/core/lib/surface/completion_queue_factory.cc b/src/core/lib/surface/completion_queue_factory.cc index 51c1183c5f..ed92dd7eba 100644 --- a/src/core/lib/surface/completion_queue_factory.cc +++ b/src/core/lib/surface/completion_queue_factory.cc @@ -30,8 +30,9 @@ static grpc_completion_queue* default_create( const grpc_completion_queue_factory* factory, const grpc_completion_queue_attributes* attr) { - return grpc_completion_queue_create_internal(attr->cq_completion_type, - attr->cq_polling_type); + return grpc_completion_queue_create_internal( + attr->cq_completion_type, attr->cq_polling_type, + static_cast(attr->cq_shutdown_cb)); } static grpc_completion_queue_factory_vtable default_vtable = {default_create}; @@ -60,14 +61,22 @@ const grpc_completion_queue_factory* grpc_completion_queue_factory_lookup( grpc_completion_queue* grpc_completion_queue_create_for_next(void* reserved) { GPR_ASSERT(!reserved); grpc_completion_queue_attributes attr = {1, GRPC_CQ_NEXT, - GRPC_CQ_DEFAULT_POLLING}; + GRPC_CQ_DEFAULT_POLLING, nullptr}; return g_default_cq_factory.vtable->create(&g_default_cq_factory, &attr); } grpc_completion_queue* grpc_completion_queue_create_for_pluck(void* reserved) { GPR_ASSERT(!reserved); grpc_completion_queue_attributes attr = {1, GRPC_CQ_PLUCK, - GRPC_CQ_DEFAULT_POLLING}; + GRPC_CQ_DEFAULT_POLLING, nullptr}; + return g_default_cq_factory.vtable->create(&g_default_cq_factory, &attr); +} + +grpc_completion_queue* grpc_completion_queue_create_for_callback( + void* shutdown_callback, void* reserved) { + GPR_ASSERT(!reserved); + grpc_completion_queue_attributes attr = { + 2, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING, shutdown_callback}; return g_default_cq_factory.vtable->create(&g_default_cq_factory, &attr); } diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m index bda1c3360b..69db340e98 100644 --- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m +++ b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m @@ -21,11 +21,11 @@ #import #ifdef GRPC_CFSTREAM -const grpc_completion_queue_attributes kCompletionQueueAttr = {GRPC_CQ_CURRENT_VERSION, - GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING}; +const grpc_completion_queue_attributes kCompletionQueueAttr = { + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING, nullptr}; #else const grpc_completion_queue_attributes kCompletionQueueAttr = { - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}; + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING, nullptr}; #endif @implementation GRPCCompletionQueue diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.c b/src/ruby/ext/grpc/rb_grpc_imports.generated.c index 38b68462df..8a2edc41f8 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.c +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.c @@ -43,6 +43,7 @@ grpc_g_stands_for_type grpc_g_stands_for_import; grpc_completion_queue_factory_lookup_type grpc_completion_queue_factory_lookup_import; grpc_completion_queue_create_for_next_type grpc_completion_queue_create_for_next_import; grpc_completion_queue_create_for_pluck_type grpc_completion_queue_create_for_pluck_import; +grpc_completion_queue_create_for_callback_type grpc_completion_queue_create_for_callback_import; grpc_completion_queue_create_type grpc_completion_queue_create_import; grpc_completion_queue_next_type grpc_completion_queue_next_import; grpc_completion_queue_pluck_type grpc_completion_queue_pluck_import; @@ -294,6 +295,7 @@ void grpc_rb_load_imports(HMODULE library) { grpc_completion_queue_factory_lookup_import = (grpc_completion_queue_factory_lookup_type) GetProcAddress(library, "grpc_completion_queue_factory_lookup"); grpc_completion_queue_create_for_next_import = (grpc_completion_queue_create_for_next_type) GetProcAddress(library, "grpc_completion_queue_create_for_next"); grpc_completion_queue_create_for_pluck_import = (grpc_completion_queue_create_for_pluck_type) GetProcAddress(library, "grpc_completion_queue_create_for_pluck"); + grpc_completion_queue_create_for_callback_import = (grpc_completion_queue_create_for_callback_type) GetProcAddress(library, "grpc_completion_queue_create_for_callback"); grpc_completion_queue_create_import = (grpc_completion_queue_create_type) GetProcAddress(library, "grpc_completion_queue_create"); grpc_completion_queue_next_import = (grpc_completion_queue_next_type) GetProcAddress(library, "grpc_completion_queue_next"); grpc_completion_queue_pluck_import = (grpc_completion_queue_pluck_type) GetProcAddress(library, "grpc_completion_queue_pluck"); diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h index d6add00d12..5a7884cdcd 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h @@ -104,6 +104,9 @@ extern grpc_completion_queue_create_for_next_type grpc_completion_queue_create_f typedef grpc_completion_queue*(*grpc_completion_queue_create_for_pluck_type)(void* reserved); extern grpc_completion_queue_create_for_pluck_type grpc_completion_queue_create_for_pluck_import; #define grpc_completion_queue_create_for_pluck grpc_completion_queue_create_for_pluck_import +typedef grpc_completion_queue*(*grpc_completion_queue_create_for_callback_type)(void* shutdown_callback, void* reserved); +extern grpc_completion_queue_create_for_callback_type grpc_completion_queue_create_for_callback_import; +#define grpc_completion_queue_create_for_callback grpc_completion_queue_create_for_callback_import typedef grpc_completion_queue*(*grpc_completion_queue_create_type)(const grpc_completion_queue_factory* factory, const grpc_completion_queue_attributes* attributes, void* reserved); extern grpc_completion_queue_create_type grpc_completion_queue_create_import; #define grpc_completion_queue_create grpc_completion_queue_create_import diff --git a/test/core/surface/public_headers_must_be_c89.c b/test/core/surface/public_headers_must_be_c89.c index 7b3e875cf0..69b3de16c4 100644 --- a/test/core/surface/public_headers_must_be_c89.c +++ b/test/core/surface/public_headers_must_be_c89.c @@ -82,6 +82,7 @@ int main(int argc, char **argv) { printf("%lx", (unsigned long) grpc_completion_queue_factory_lookup); printf("%lx", (unsigned long) grpc_completion_queue_create_for_next); printf("%lx", (unsigned long) grpc_completion_queue_create_for_pluck); + printf("%lx", (unsigned long) grpc_completion_queue_create_for_callback); printf("%lx", (unsigned long) grpc_completion_queue_create); printf("%lx", (unsigned long) grpc_completion_queue_next); printf("%lx", (unsigned long) grpc_completion_queue_pluck); -- cgit v1.2.3 From a59e48e889f6afd6f915fce5d0638f0a3697df06 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 14 Aug 2018 02:18:32 -0700 Subject: Add a test of callback CQ --- test/core/surface/completion_queue_test.cc | 80 +++++++++++++++++++++++++++++- 1 file changed, 79 insertions(+), 1 deletion(-) (limited to 'test/core/surface') diff --git a/test/core/surface/completion_queue_test.cc b/test/core/surface/completion_queue_test.cc index 68129146cc..b889fd0fc6 100644 --- a/test/core/surface/completion_queue_test.cc +++ b/test/core/surface/completion_queue_test.cc @@ -22,6 +22,7 @@ #include #include #include "src/core/lib/gpr/useful.h" +#include "src/core/lib/gprpp/memory.h" #include "src/core/lib/iomgr/iomgr.h" #include "test/core/util/test_config.h" @@ -41,11 +42,18 @@ static void shutdown_and_destroy(grpc_completion_queue* cc) { case GRPC_CQ_NEXT: { ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), nullptr); + GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN); break; } case GRPC_CQ_PLUCK: { ev = grpc_completion_queue_pluck( cc, create_test_tag(), gpr_inf_past(GPR_CLOCK_REALTIME), nullptr); + GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN); + break; + } + case GRPC_CQ_CALLBACK: { + // Nothing to do here. The shutdown callback will be invoked when + // possible. break; } default: { @@ -54,7 +62,6 @@ static void shutdown_and_destroy(grpc_completion_queue* cc) { } } - GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN); grpc_completion_queue_destroy(cc); } @@ -350,6 +357,76 @@ static void test_pluck_after_shutdown(void) { } } +static void test_callback(void) { + grpc_completion_queue* cc; + void* tags[128]; + grpc_cq_completion completions[GPR_ARRAY_SIZE(tags)]; + grpc_cq_polling_type polling_types[] = { + GRPC_CQ_DEFAULT_POLLING, GRPC_CQ_NON_LISTENING, GRPC_CQ_NON_POLLING}; + grpc_completion_queue_attributes attr; + unsigned i; + + LOG_TEST("test_callback"); + + bool got_shutdown = false; + class ShutdownCallback : public grpc_core::CQCallbackInterface { + public: + ShutdownCallback(bool* done) : done_(done) {} + ~ShutdownCallback() {} + void Run(bool ok) override { *done_ = ok; } + + private: + bool* done_; + }; + ShutdownCallback shutdown_cb(&got_shutdown); + + attr.version = 2; + attr.cq_completion_type = GRPC_CQ_CALLBACK; + attr.cq_shutdown_cb = &shutdown_cb; + + for (size_t pidx = 0; pidx < GPR_ARRAY_SIZE(polling_types); pidx++) { + grpc_core::ExecCtx exec_ctx; // reset exec_ctx + attr.cq_polling_type = polling_types[pidx]; + cc = grpc_completion_queue_create( + grpc_completion_queue_factory_lookup(&attr), &attr, nullptr); + + int counter = 0; + class TagCallback : public grpc_core::CQCallbackInterface { + public: + TagCallback(int* counter, int tag) : counter_(counter), tag_(tag) {} + ~TagCallback() {} + void Run(bool ok) override { + GPR_ASSERT(ok); + *counter_ += tag_; + grpc_core::Delete(this); + }; + + private: + int* counter_; + int tag_; + }; + + int sumtags = 0; + for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { + tags[i] = static_cast(grpc_core::New(&counter, i)); + sumtags += i; + } + + for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { + GPR_ASSERT(grpc_cq_begin_op(cc, tags[i])); + grpc_cq_end_op(cc, tags[i], GRPC_ERROR_NONE, do_nothing_end_completion, + nullptr, &completions[i]); + } + + GPR_ASSERT(sumtags == counter); + + shutdown_and_destroy(cc); + + GPR_ASSERT(got_shutdown); + got_shutdown = false; + } +} + struct thread_state { grpc_completion_queue* cc; void* tag; @@ -368,6 +445,7 @@ int main(int argc, char** argv) { test_pluck_after_shutdown(); test_cq_tls_cache_full(); test_cq_tls_cache_empty(); + test_callback(); grpc_shutdown(); return 0; } -- cgit v1.2.3 From b59d8674d24d9b40d8d8b2b40f2be0118ff524d2 Mon Sep 17 00:00:00 2001 From: Eric Gribkoff Date: Fri, 10 Aug 2018 10:13:36 -0700 Subject: Python post-fork handler: exit if grpc shutdown fails --- grpc.def | 1 + include/grpc/grpc.h | 6 ++++++ src/core/lib/iomgr/fork_posix.cc | 2 +- src/core/lib/surface/init.h | 1 - src/python/grpcio/grpc/_cython/_cygrpc/fork_posix.pyx.pxi | 4 ++++ src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi | 1 + src/ruby/ext/grpc/rb_grpc_imports.generated.c | 2 ++ src/ruby/ext/grpc/rb_grpc_imports.generated.h | 3 +++ test/core/surface/public_headers_must_be_c89.c | 1 + 9 files changed, 19 insertions(+), 2 deletions(-) (limited to 'test/core/surface') diff --git a/grpc.def b/grpc.def index 009de4e868..72e3e90c62 100644 --- a/grpc.def +++ b/grpc.def @@ -15,6 +15,7 @@ EXPORTS grpc_register_plugin grpc_init grpc_shutdown + grpc_is_initialized grpc_version_string grpc_g_stands_for grpc_completion_queue_factory_lookup diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index 897b89851a..4e50cd0bac 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -79,6 +79,12 @@ GRPCAPI void grpc_init(void); destroyed. */ GRPCAPI void grpc_shutdown(void); +/** EXPERIMENTAL. Returns 1 if the grpc library has been initialized. + TODO(ericgribkoff) Decide if this should be promoted to non-experimental as + part of stabilizing the fork support API, as tracked in + https://github.com/grpc/grpc/issues/15334 */ +GRPCAPI int grpc_is_initialized(void); + /** Return a string representing the current version of grpc */ GRPCAPI const char* grpc_version_string(void); diff --git a/src/core/lib/iomgr/fork_posix.cc b/src/core/lib/iomgr/fork_posix.cc index a5b61fb4ce..ac85c81de2 100644 --- a/src/core/lib/iomgr/fork_posix.cc +++ b/src/core/lib/iomgr/fork_posix.cc @@ -25,6 +25,7 @@ #include #include +#include #include #include "src/core/lib/gpr/env.h" @@ -34,7 +35,6 @@ #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/timer_manager.h" #include "src/core/lib/iomgr/wakeup_fd_posix.h" -#include "src/core/lib/surface/init.h" /* * NOTE: FORKING IS NOT GENERALLY SUPPORTED, THIS IS ONLY INTENDED TO WORK diff --git a/src/core/lib/surface/init.h b/src/core/lib/surface/init.h index 9353208332..193f51447d 100644 --- a/src/core/lib/surface/init.h +++ b/src/core/lib/surface/init.h @@ -22,6 +22,5 @@ void grpc_register_security_filters(void); void grpc_security_pre_init(void); void grpc_security_init(void); -int grpc_is_initialized(void); #endif /* GRPC_CORE_LIB_SURFACE_INIT_H */ diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/fork_posix.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/fork_posix.pyx.pxi index 1176258da8..0d2516977b 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/fork_posix.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/fork_posix.pyx.pxi @@ -73,6 +73,10 @@ cdef void __postfork_child() nogil: # TODO(ericgribkoff) Check and abort if core is not shutdown with _fork_state.fork_in_progress_condition: _fork_state.fork_in_progress = False + if grpc_is_initialized() > 0: + with gil: + _LOGGER.error('Failed to shutdown gRPC Core after fork()') + os._exit(os.EX_USAGE) def fork_handlers_and_grpc_init(): diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi index bcbfec0c9f..4781219319 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi @@ -322,6 +322,7 @@ cdef extern from "grpc/grpc.h": void grpc_init() nogil void grpc_shutdown() nogil + int grpc_is_initialized() nogil ctypedef struct grpc_completion_queue_factory: pass diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.c b/src/ruby/ext/grpc/rb_grpc_imports.generated.c index 37b97aabd4..f44f89dfa0 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.c +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.c @@ -38,6 +38,7 @@ grpc_call_details_destroy_type grpc_call_details_destroy_import; grpc_register_plugin_type grpc_register_plugin_import; grpc_init_type grpc_init_import; grpc_shutdown_type grpc_shutdown_import; +grpc_is_initialized_type grpc_is_initialized_import; grpc_version_string_type grpc_version_string_import; grpc_g_stands_for_type grpc_g_stands_for_import; grpc_completion_queue_factory_lookup_type grpc_completion_queue_factory_lookup_import; @@ -291,6 +292,7 @@ void grpc_rb_load_imports(HMODULE library) { grpc_register_plugin_import = (grpc_register_plugin_type) GetProcAddress(library, "grpc_register_plugin"); grpc_init_import = (grpc_init_type) GetProcAddress(library, "grpc_init"); grpc_shutdown_import = (grpc_shutdown_type) GetProcAddress(library, "grpc_shutdown"); + grpc_is_initialized_import = (grpc_is_initialized_type) GetProcAddress(library, "grpc_is_initialized"); grpc_version_string_import = (grpc_version_string_type) GetProcAddress(library, "grpc_version_string"); grpc_g_stands_for_import = (grpc_g_stands_for_type) GetProcAddress(library, "grpc_g_stands_for"); grpc_completion_queue_factory_lookup_import = (grpc_completion_queue_factory_lookup_type) GetProcAddress(library, "grpc_completion_queue_factory_lookup"); diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h index f7a00046e3..99433f0e40 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h @@ -89,6 +89,9 @@ extern grpc_init_type grpc_init_import; typedef void(*grpc_shutdown_type)(void); extern grpc_shutdown_type grpc_shutdown_import; #define grpc_shutdown grpc_shutdown_import +typedef int(*grpc_is_initialized_type)(void); +extern grpc_is_initialized_type grpc_is_initialized_import; +#define grpc_is_initialized grpc_is_initialized_import typedef const char*(*grpc_version_string_type)(void); extern grpc_version_string_type grpc_version_string_import; #define grpc_version_string grpc_version_string_import diff --git a/test/core/surface/public_headers_must_be_c89.c b/test/core/surface/public_headers_must_be_c89.c index 42dc95b9bb..b832a1661b 100644 --- a/test/core/surface/public_headers_must_be_c89.c +++ b/test/core/surface/public_headers_must_be_c89.c @@ -77,6 +77,7 @@ int main(int argc, char **argv) { printf("%lx", (unsigned long) grpc_register_plugin); printf("%lx", (unsigned long) grpc_init); printf("%lx", (unsigned long) grpc_shutdown); + printf("%lx", (unsigned long) grpc_is_initialized); printf("%lx", (unsigned long) grpc_version_string); printf("%lx", (unsigned long) grpc_g_stands_for); printf("%lx", (unsigned long) grpc_completion_queue_factory_lookup); -- cgit v1.2.3