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(-) 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 e509e70b2da20fe837efdca0fe8c550ced1128a9 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Fri, 10 Aug 2018 15:01:25 -0700 Subject: Obj-c doesn't have nullptr, use NULL --- src/objective-c/GRPCClient/private/GRPCCompletionQueue.m | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m index 69db340e98..1e2537a5b1 100644 --- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m +++ b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m @@ -22,10 +22,10 @@ #ifdef GRPC_CFSTREAM const grpc_completion_queue_attributes kCompletionQueueAttr = { - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING, nullptr}; + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING, NULL}; #else const grpc_completion_queue_attributes kCompletionQueueAttr = { - GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING, nullptr}; + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING, NULL}; #endif @implementation GRPCCompletionQueue -- cgit v1.2.3 From fe7f79189be8c774a3cc210d4b86191179742ca6 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 14 Aug 2018 00:58:33 -0700 Subject: Address reviewer comments --- src/core/lib/surface/completion_queue.cc | 39 ++++++++++++++++++++------------ src/core/lib/surface/completion_queue.h | 9 ++++++-- 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/src/core/lib/surface/completion_queue.cc b/src/core/lib/surface/completion_queue.cc index fd33ce044c..9086578f7c 100644 --- a/src/core/lib/surface/completion_queue.cc +++ b/src/core/lib/surface/completion_queue.cc @@ -263,12 +263,6 @@ typedef struct cq_callback_data { 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; @@ -308,6 +302,12 @@ 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); +// A cq_end_op function is called when an operation on a given CQ with +// a given tag has completed. The storage argument is a reference to the +// space reserved for this completion as it is placed into the corresponding +// queue. The done argument is a callback that will be invoked when it is +// safe to free up that storage. The storage MUST NOT be freed until the +// done callback is invoked. static void cq_end_op_for_next(grpc_completion_queue* cq, void* tag, grpc_error* error, void (*done)(void* done_arg, @@ -332,8 +332,11 @@ static grpc_event cq_next(grpc_completion_queue* cq, gpr_timespec deadline, static grpc_event cq_pluck(grpc_completion_queue* cq, void* tag, gpr_timespec deadline, void* reserved); -static void cq_init_next(void* data, grpc_core::CQCallbackInterface*); -static void cq_init_pluck(void* data, grpc_core::CQCallbackInterface*); +// Note that cq_init_next and cq_init_pluck do not use the shutdown_callback +static void cq_init_next(void* data, + grpc_core::CQCallbackInterface* shutdown_callback); +static void cq_init_pluck(void* data, + grpc_core::CQCallbackInterface* shutdown_callback); static void cq_init_callback(void* data, grpc_core::CQCallbackInterface* shutdown_callback); static void cq_destroy_next(void* data); @@ -494,7 +497,11 @@ grpc_completion_queue* grpc_completion_queue_create_internal( return cq; } -static void cq_init_next(void* ptr, grpc_core::CQCallbackInterface*) { +static void cq_init_next(void* ptr, + grpc_core::CQCallbackInterface* shutdown_callback) { + // shutdown_callback should not be provided to this CQ variant + GPR_ASSERT(shutdown_callback == nullptr); + 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); @@ -509,7 +516,11 @@ static void cq_destroy_next(void* ptr) { cq_event_queue_destroy(&cqd->queue); } -static void cq_init_pluck(void* ptr, grpc_core::CQCallbackInterface*) { +static void cq_init_pluck(void* ptr, + grpc_core::CQCallbackInterface* shutdown_callback) { + // shutdown_callback should not be provided to this CQ variant + GPR_ASSERT(shutdown_callback == nullptr); + 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); @@ -531,7 +542,6 @@ static void cq_init_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; @@ -838,7 +848,8 @@ static void cq_end_op_for_callback( } } - /* We don't care for the storage content */ + // The callback-based CQ isn't really a queue at all and thus has no need + // for reserved storage. Invoke the done callback right away to release it. done(done_arg, storage); gpr_mu_lock(cq->mu); @@ -1336,8 +1347,6 @@ static void cq_finish_shutdown_callback(grpc_completion_queue* 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); @@ -1347,7 +1356,7 @@ 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. + * We call cq_finish_shutdown_callback() below, which calls 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 diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 6d8c6c9b06..5aa54682e0 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -47,8 +47,13 @@ typedef struct grpc_cq_completion { uintptr_t next; } grpc_cq_completion; -/// For callback CQs, the following is what is actually intended by -/// the tag. +/// For callback CQs, the tag that is passed in for an operation must +/// actually be a pointer to an implementation of the following class. +/// When the operation completes, the tag will be typecasted from void* +/// to grpc_core::CQCallbackInterface* and then the Run method will be +/// invoked on it. In practice, the language binding (e.g., C++ API +/// implementation) is responsible for providing and using an implementation +/// of this abstract base class. namespace grpc_core { class CQCallbackInterface { public: -- cgit v1.2.3 From d0f116c885af5b7b4570373c33f6960108bc8baa Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 14 Aug 2018 01:43:56 -0700 Subject: Can't count on shutdown_cb nullptr if version is 1 --- src/core/lib/surface/completion_queue.cc | 6 ------ 1 file changed, 6 deletions(-) diff --git a/src/core/lib/surface/completion_queue.cc b/src/core/lib/surface/completion_queue.cc index 9086578f7c..3ded712b70 100644 --- a/src/core/lib/surface/completion_queue.cc +++ b/src/core/lib/surface/completion_queue.cc @@ -499,9 +499,6 @@ grpc_completion_queue* grpc_completion_queue_create_internal( static void cq_init_next(void* ptr, grpc_core::CQCallbackInterface* shutdown_callback) { - // shutdown_callback should not be provided to this CQ variant - GPR_ASSERT(shutdown_callback == nullptr); - 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); @@ -518,9 +515,6 @@ static void cq_destroy_next(void* ptr) { static void cq_init_pluck(void* ptr, grpc_core::CQCallbackInterface* shutdown_callback) { - // shutdown_callback should not be provided to this CQ variant - GPR_ASSERT(shutdown_callback == nullptr); - 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); -- cgit v1.2.3 From e91ae9d69489d7b031aa09240ba6f044b2692ad6 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 14 Aug 2018 02:18:15 -0700 Subject: Fix abstract base class definition --- src/core/lib/surface/completion_queue.h | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 5aa54682e0..a7c524d8e8 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -25,6 +25,7 @@ #include #include "src/core/lib/debug/trace.h" +#include "src/core/lib/gprpp/abstract.h" #include "src/core/lib/iomgr/pollset.h" /* These trace flags default to 1. The corresponding lines are only traced @@ -58,7 +59,9 @@ namespace grpc_core { class CQCallbackInterface { public: virtual ~CQCallbackInterface() {} - virtual void Run(bool) = 0; + virtual void Run(bool) GRPC_ABSTRACT; + + GRPC_ABSTRACT_BASE_CLASS }; } // namespace grpc_core -- 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(-) 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