diff --git a/include/grpc++/server_context.h b/include/grpc++/server_context.h index cf2732b33d2..23273f43e67 100644 --- a/include/grpc++/server_context.h +++ b/include/grpc++/server_context.h @@ -125,6 +125,14 @@ class ServerContext { const struct census_context* census_context() const; + // Async only. Has to be called before the rpc starts. + // Returns the tag in completion queue when the rpc finishes. + // IsCancelled() can then be called to check whether the rpc was cancelled. + void AsyncNotifyWhenDone(void* tag) { + has_notify_when_done_tag_ = true; + async_notify_when_done_tag_ = tag; + } + private: friend class ::grpc::testing::InteropContextInspector; friend class ::grpc::Server; @@ -165,6 +173,8 @@ class ServerContext { void set_call(grpc_call* call); CompletionOp* completion_op_; + bool has_notify_when_done_tag_; + void* async_notify_when_done_tag_; gpr_timespec deadline_; grpc_call* call_; diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index b928f123b5a..407f36ad82e 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -589,7 +589,7 @@ void grpc_server_register_completion_queue(grpc_server *server, /** Add a HTTP2 over plaintext over tcp listener. Returns bound port number on success, 0 on failure. REQUIRES: server not started */ -int grpc_server_add_http2_port(grpc_server *server, const char *addr); +int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr); /** Start a server - tells all listeners to start listening */ void grpc_server_start(grpc_server *server); diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c index 78c53466b38..4ab845bc008 100644 --- a/src/core/surface/server_chttp2.c +++ b/src/core/surface/server_chttp2.c @@ -80,7 +80,7 @@ static void destroy(grpc_server *server, void *tcpp) { grpc_tcp_server_destroy(tcp, grpc_server_listener_destroy_done, server); } -int grpc_server_add_http2_port(grpc_server *server, const char *addr) { +int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr) { grpc_resolved_addresses *resolved = NULL; grpc_tcp_server *tcp = NULL; size_t i; diff --git a/src/cpp/server/insecure_server_credentials.cc b/src/cpp/server/insecure_server_credentials.cc index aca3568e597..800cd36caaf 100644 --- a/src/cpp/server/insecure_server_credentials.cc +++ b/src/cpp/server/insecure_server_credentials.cc @@ -41,7 +41,7 @@ class InsecureServerCredentialsImpl GRPC_FINAL : public ServerCredentials { public: int AddPortToServer(const grpc::string& addr, grpc_server* server) GRPC_OVERRIDE { - return grpc_server_add_http2_port(server, addr.c_str()); + return grpc_server_add_insecure_http2_port(server, addr.c_str()); } }; } // namespace diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index cf19556e7a0..04373397f92 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -50,16 +50,23 @@ namespace grpc { class ServerContext::CompletionOp GRPC_FINAL : public CallOpSetInterface { public: // initial refs: one in the server context, one in the cq - CompletionOp() : refs_(2), finalized_(false), cancelled_(0) {} + CompletionOp() : has_tag_(false), tag_(nullptr), refs_(2), finalized_(false), cancelled_(0) {} void FillOps(grpc_op* ops, size_t* nops) GRPC_OVERRIDE; bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE; bool CheckCancelled(CompletionQueue* cq); + void set_tag(void* tag) { + has_tag_ = true; + tag_ = tag; + } + void Unref(); private: + bool has_tag_; + void* tag_; grpc::mutex mu_; int refs_; bool finalized_; @@ -90,18 +97,25 @@ void ServerContext::CompletionOp::FillOps(grpc_op* ops, size_t* nops) { bool ServerContext::CompletionOp::FinalizeResult(void** tag, bool* status) { grpc::unique_lock lock(mu_); finalized_ = true; + bool ret = false; + if (has_tag_) { + *tag = tag_; + ret = true; + } if (!*status) cancelled_ = 1; if (--refs_ == 0) { lock.unlock(); delete this; } - return false; + return ret; } // ServerContext body ServerContext::ServerContext() : completion_op_(nullptr), + has_notify_when_done_tag_(false), + async_notify_when_done_tag_(nullptr), call_(nullptr), cq_(nullptr), sent_initial_metadata_(false) {} @@ -109,6 +123,8 @@ ServerContext::ServerContext() ServerContext::ServerContext(gpr_timespec deadline, grpc_metadata* metadata, size_t metadata_count) : completion_op_(nullptr), + has_notify_when_done_tag_(false), + async_notify_when_done_tag_(nullptr), deadline_(deadline), call_(nullptr), cq_(nullptr), @@ -133,6 +149,9 @@ ServerContext::~ServerContext() { void ServerContext::BeginCompletionOp(Call* call) { GPR_ASSERT(!completion_op_); completion_op_ = new CompletionOp(); + if (has_notify_when_done_tag_) { + completion_op_->set_tag(async_notify_when_done_tag_); + } call->PerformOps(completion_op_); } diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c index 65f33030552..c12b6361ec3 100644 --- a/src/csharp/ext/grpc_csharp_ext.c +++ b/src/csharp/ext/grpc_csharp_ext.c @@ -732,7 +732,7 @@ grpcsharp_server_create(grpc_completion_queue *cq, GPR_EXPORT gpr_int32 GPR_CALLTYPE grpcsharp_server_add_insecure_http2_port(grpc_server *server, const char *addr) { - return grpc_server_add_http2_port(server, addr); + return grpc_server_add_insecure_http2_port(server, addr); } GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_start(grpc_server *server) { diff --git a/src/node/ext/server.cc b/src/node/ext/server.cc index 04fabc871d2..1dc179db3d7 100644 --- a/src/node/ext/server.cc +++ b/src/node/ext/server.cc @@ -265,8 +265,8 @@ NAN_METHOD(Server::AddHttp2Port) { grpc_server_credentials *creds = creds_object->GetWrappedServerCredentials(); int port; if (creds == NULL) { - port = grpc_server_add_http2_port(server->wrapped_server, - *NanUtf8String(args[0])); + port = grpc_server_add_insecure_http2_port(server->wrapped_server, + *NanUtf8String(args[0])); } else { port = grpc_server_add_secure_http2_port(server->wrapped_server, *NanUtf8String(args[0]), diff --git a/src/php/ext/grpc/server.c b/src/php/ext/grpc/server.c index 8b8d5b2f476..d58aa884ca8 100644 --- a/src/php/ext/grpc/server.c +++ b/src/php/ext/grpc/server.c @@ -182,7 +182,7 @@ PHP_METHOD(Server, addHttp2Port) { "add_http2_port expects a string", 1 TSRMLS_CC); return; } - RETURN_LONG(grpc_server_add_http2_port(server->wrapped, addr)); + RETURN_LONG(grpc_server_add_insecure_http2_port(server->wrapped, addr)); } PHP_METHOD(Server, addSecureHttp2Port) { diff --git a/src/python/grpcio/grpc/_adapter/_c/types/server.c b/src/python/grpcio/grpc/_adapter/_c/types/server.c index 2a00f34039c..c2190ea672d 100644 --- a/src/python/grpcio/grpc/_adapter/_c/types/server.c +++ b/src/python/grpcio/grpc/_adapter/_c/types/server.c @@ -155,7 +155,7 @@ PyObject *pygrpc_Server_add_http2_port( port = grpc_server_add_secure_http2_port( self->c_serv, addr, creds->c_creds); } else { - port = grpc_server_add_http2_port(self->c_serv, addr); + port = grpc_server_add_insecure_http2_port(self->c_serv, addr); } return PyInt_FromLong(port); diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c index 375a651d247..79a4ae8757b 100644 --- a/src/ruby/ext/grpc/rb_server.c +++ b/src/ruby/ext/grpc/rb_server.c @@ -357,7 +357,8 @@ static VALUE grpc_rb_server_add_http2_port(int argc, VALUE *argv, VALUE self) { rb_raise(rb_eRuntimeError, "destroyed!"); return Qnil; } else if (rb_creds == Qnil) { - recvd_port = grpc_server_add_http2_port(s->wrapped, StringValueCStr(port)); + recvd_port = + grpc_server_add_insecure_http2_port(s->wrapped, StringValueCStr(port)); if (recvd_port == 0) { rb_raise(rb_eRuntimeError, "could not add port %s to server, not sure why", diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c index 3c227b7e295..48a88be5255 100644 --- a/test/core/end2end/dualstack_socket_test.c +++ b/test/core/end2end/dualstack_socket_test.c @@ -96,8 +96,8 @@ void test_connect(const char *server_host, const char *client_host, int port, cq = grpc_completion_queue_create(); server = grpc_server_create(NULL); grpc_server_register_completion_queue(server, cq); - GPR_ASSERT((got_port = grpc_server_add_http2_port(server, server_hostport)) > - 0); + GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port( + server, server_hostport)) > 0); if (port == 0) { port = got_port; } else { diff --git a/test/core/end2end/fixtures/chttp2_fullstack.c b/test/core/end2end/fixtures/chttp2_fullstack.c index 6647b949ba2..53a6f0d7a57 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack.c +++ b/test/core/end2end/fixtures/chttp2_fullstack.c @@ -84,7 +84,7 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f, } f->server = grpc_server_create(server_args); grpc_server_register_completion_queue(f->server, f->cq); - GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); + GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_fullstack_compression.c b/test/core/end2end/fixtures/chttp2_fullstack_compression.c index 7eb247b5409..a75075d1d4f 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack_compression.c +++ b/test/core/end2end/fixtures/chttp2_fullstack_compression.c @@ -99,7 +99,7 @@ void chttp2_init_server_fullstack_compression(grpc_end2end_test_fixture *f, } f->server = grpc_server_create(ffd->server_args_compression); grpc_server_register_completion_queue(f->server, f->cq); - GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); + GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c index 89ad7b8c2d8..20afdb868e9 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c +++ b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c @@ -89,7 +89,7 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f, } f->server = grpc_server_create(server_args); grpc_server_register_completion_queue(f->server, f->cq); - GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); + GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c index a2ab25d886f..8491ea6970d 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c +++ b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c @@ -89,7 +89,7 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f, } f->server = grpc_server_create(server_args); grpc_server_register_completion_queue(f->server, f->cq); - GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); + GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c b/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c index 93786d0943a..2a4835add12 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c +++ b/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c @@ -83,7 +83,7 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f, } f->server = grpc_server_create(server_args); grpc_server_register_completion_queue(f->server, f->cq); - GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); + GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } diff --git a/test/core/end2end/multiple_server_queues_test.c b/test/core/end2end/multiple_server_queues_test.c index 208d42e6e74..7772d14ba5f 100644 --- a/test/core/end2end/multiple_server_queues_test.c +++ b/test/core/end2end/multiple_server_queues_test.c @@ -45,7 +45,7 @@ int main(int argc, char **argv) { cq2 = grpc_completion_queue_create(); server = grpc_server_create(NULL); grpc_server_register_completion_queue(server, cq1); - grpc_server_add_http2_port(server, "[::]:0"); + grpc_server_add_insecure_http2_port(server, "[::]:0"); grpc_server_register_completion_queue(server, cq2); grpc_server_start(server); grpc_server_shutdown_and_notify(server, cq2, NULL); diff --git a/test/core/fling/server.c b/test/core/fling/server.c index 8f349044d97..f445c681782 100644 --- a/test/core/fling/server.c +++ b/test/core/fling/server.c @@ -216,7 +216,7 @@ int main(int argc, char **argv) { grpc_server_credentials_release(ssl_creds); } else { server = grpc_server_create(NULL); - GPR_ASSERT(grpc_server_add_http2_port(server, addr)); + GPR_ASSERT(grpc_server_add_insecure_http2_port(server, addr)); } grpc_server_register_completion_queue(server, cq); grpc_server_start(server); diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index b95bdf6b9b4..9b53bdc9990 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -592,6 +592,80 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { EXPECT_EQ(meta6.second, server_trailing_metadata.find(meta6.first)->second); EXPECT_GE(server_trailing_metadata.size(), static_cast(2)); } + +// Server uses AsyncNotifyWhenDone API to check for cancellation +TEST_F(AsyncEnd2endTest, ServerCheckCancellation) { + ResetStub(); + + EchoRequest send_request; + EchoRequest recv_request; + EchoResponse send_response; + EchoResponse recv_response; + Status recv_status; + + ClientContext cli_ctx; + ServerContext srv_ctx; + grpc::ServerAsyncResponseWriter response_writer(&srv_ctx); + + send_request.set_message("Hello"); + std::unique_ptr > response_reader( + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); + + srv_ctx.AsyncNotifyWhenDone(tag(5)); + service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(), + cq_.get(), tag(2)); + + Verifier().Expect(2, true).Verify(cq_.get()); + EXPECT_EQ(send_request.message(), recv_request.message()); + + cli_ctx.TryCancel(); + Verifier().Expect(5, true).Verify(cq_.get()); + EXPECT_TRUE(srv_ctx.IsCancelled()); + + response_reader->Finish(&recv_response, &recv_status, tag(4)); + Verifier().Expect(4, false).Verify(cq_.get()); + + EXPECT_EQ(StatusCode::CANCELLED, recv_status.error_code()); +} + +// Server uses AsyncNotifyWhenDone API to check for normal finish +TEST_F(AsyncEnd2endTest, ServerCheckDone) { + ResetStub(); + + EchoRequest send_request; + EchoRequest recv_request; + EchoResponse send_response; + EchoResponse recv_response; + Status recv_status; + + ClientContext cli_ctx; + ServerContext srv_ctx; + grpc::ServerAsyncResponseWriter response_writer(&srv_ctx); + + send_request.set_message("Hello"); + std::unique_ptr > response_reader( + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); + + srv_ctx.AsyncNotifyWhenDone(tag(5)); + service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(), + cq_.get(), tag(2)); + + Verifier().Expect(2, true).Verify(cq_.get()); + EXPECT_EQ(send_request.message(), recv_request.message()); + + send_response.set_message(recv_request.message()); + response_writer.Finish(send_response, Status::OK, tag(3)); + Verifier().Expect(3, true).Verify(cq_.get()); + Verifier().Expect(5, true).Verify(cq_.get()); + EXPECT_FALSE(srv_ctx.IsCancelled()); + + response_reader->Finish(&recv_response, &recv_status, tag(4)); + Verifier().Expect(4, true).Verify(cq_.get()); + + EXPECT_EQ(send_response.message(), recv_response.message()); + EXPECT_TRUE(recv_status.ok()); +} + } // namespace } // namespace testing } // namespace grpc