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