From a324c4fea63840eb06dadb4f0d920226b034404e Mon Sep 17 00:00:00 2001 From: yang-g Date: Fri, 24 Jun 2016 11:57:07 -0700 Subject: Add API to get c slice from c++ Slice. --- include/grpc++/support/slice.h | 3 +++ 1 file changed, 3 insertions(+) (limited to 'include') diff --git a/include/grpc++/support/slice.h b/include/grpc++/support/slice.h index cec9062d4f..5874b4f5ae 100644 --- a/include/grpc++/support/slice.h +++ b/include/grpc++/support/slice.h @@ -77,6 +77,9 @@ class Slice GRPC_FINAL { /// Raw pointer to the end (one byte \em past the last element) of the slice. const uint8_t* end() const { return GPR_SLICE_END_PTR(slice_); } + /// Raw C slice. Caller needs to call gpr_slice_unref when done. + gpr_slice c_slice() const { return gpr_slice_ref(slice_); } + private: friend class ByteBuffer; -- cgit v1.2.3 From e89dc6ceb60669e97b74e52311497af110dc4f21 Mon Sep 17 00:00:00 2001 From: yang-g Date: Mon, 11 Jul 2016 15:48:01 -0700 Subject: Make Server::Wait work for async only server. --- include/grpc++/server.h | 2 ++ src/cpp/server/server.cc | 5 ++--- test/cpp/end2end/async_end2end_test.cc | 18 ++++++++++++++++++ 3 files changed, 22 insertions(+), 3 deletions(-) (limited to 'include') diff --git a/include/grpc++/server.h b/include/grpc++/server.h index 7a8858ef19..9a2f8f11c4 100644 --- a/include/grpc++/server.h +++ b/include/grpc++/server.h @@ -183,6 +183,8 @@ class Server GRPC_FINAL : public ServerInterface, private GrpcLibraryCodegen { int num_running_cb_; grpc::condition_variable callback_cv_; + grpc::condition_variable shutdown_cv_; + std::shared_ptr global_callbacks_; std::list* sync_methods_; diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc index fb4c68ebe4..374c9cbc04 100644 --- a/src/cpp/server/server.cc +++ b/src/cpp/server/server.cc @@ -462,14 +462,13 @@ void Server::ShutdownInternal(gpr_timespec deadline) { while (num_running_cb_ != 0) { callback_cv_.wait(lock); } + shutdown_cv_.notify_all(); } } void Server::Wait() { grpc::unique_lock lock(mu_); - while (num_running_cb_ != 0) { - callback_cv_.wait(lock); - } + shutdown_cv_.wait(lock); } void Server::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) { diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 6c7eae53a4..01ba8962d7 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -345,6 +345,24 @@ TEST_P(AsyncEnd2endTest, SequentialRpcs) { SendRpc(10); } +// We do not need to protect notify because the use is synchronized. +void ServerWait(Server* server, int* notify) { + server->Wait(); + *notify = 1; +} +TEST_P(AsyncEnd2endTest, WaitAndShutdownTest) { + int notify = 0; + std::thread* wait_thread = + new std::thread(&ServerWait, server_.get(), ¬ify); + ResetStub(); + SendRpc(1); + EXPECT_EQ(0, notify); + server_->Shutdown(); + wait_thread->join(); + EXPECT_EQ(1, notify); + delete wait_thread; +} + // Test a simple RPC using the async version of Next TEST_P(AsyncEnd2endTest, AsyncNextRpc) { ResetStub(); -- cgit v1.2.3 From 6ec11f2acd1311384cd17ea933e7bb9140ef75d7 Mon Sep 17 00:00:00 2001 From: yang-g Date: Thu, 14 Jul 2016 14:53:35 -0700 Subject: prevent spurious wake up and unstarted/already shutdown server --- include/grpc++/server.h | 1 + src/cpp/server/server.cc | 7 ++++++- test/cpp/end2end/async_end2end_test.cc | 7 +++++++ 3 files changed, 14 insertions(+), 1 deletion(-) (limited to 'include') diff --git a/include/grpc++/server.h b/include/grpc++/server.h index 9a2f8f11c4..6876961e21 100644 --- a/include/grpc++/server.h +++ b/include/grpc++/server.h @@ -179,6 +179,7 @@ class Server GRPC_FINAL : public ServerInterface, private GrpcLibraryCodegen { grpc::mutex mu_; bool started_; bool shutdown_; + bool shutdown_notified_; // The number of threads which are running callbacks. int num_running_cb_; grpc::condition_variable callback_cv_; diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc index 374c9cbc04..af04fd4ca6 100644 --- a/src/cpp/server/server.cc +++ b/src/cpp/server/server.cc @@ -281,6 +281,7 @@ Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned, : max_message_size_(max_message_size), started_(false), shutdown_(false), + shutdown_notified_(false), num_running_cb_(0), sync_methods_(new std::list), has_generic_service_(false), @@ -462,13 +463,17 @@ void Server::ShutdownInternal(gpr_timespec deadline) { while (num_running_cb_ != 0) { callback_cv_.wait(lock); } + + shutdown_notified_ = true; shutdown_cv_.notify_all(); } } void Server::Wait() { grpc::unique_lock lock(mu_); - shutdown_cv_.wait(lock); + while (started_ && !shutdown_notified_) { + shutdown_cv_.wait(lock); + } } void Server::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) { diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 01ba8962d7..4a8936d281 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -363,6 +363,13 @@ TEST_P(AsyncEnd2endTest, WaitAndShutdownTest) { delete wait_thread; } +TEST_P(AsyncEnd2endTest, ShutdownThenWait) { + ResetStub(); + SendRpc(1); + server_->Shutdown(); + server_->Wait(); +} + // Test a simple RPC using the async version of Next TEST_P(AsyncEnd2endTest, AsyncNextRpc) { ResetStub(); -- cgit v1.2.3