diff --git a/Makefile b/Makefile index 1ab1fdaefcd..4918fbc19af 100644 --- a/Makefile +++ b/Makefile @@ -2624,7 +2624,6 @@ LIBGRPC++_SRC = \ src/cpp/server/server_credentials.cc \ src/cpp/server/server_rpc_handler.cc \ src/cpp/server/thread_pool.cc \ - src/cpp/stream/stream_context.cc \ src/cpp/util/status.cc \ src/cpp/util/time.cc \ @@ -2680,7 +2679,6 @@ src/cpp/server/server_context_impl.cc: $(OPENSSL_DEP) src/cpp/server/server_credentials.cc: $(OPENSSL_DEP) src/cpp/server/server_rpc_handler.cc: $(OPENSSL_DEP) src/cpp/server/thread_pool.cc: $(OPENSSL_DEP) -src/cpp/stream/stream_context.cc: $(OPENSSL_DEP) src/cpp/util/status.cc: $(OPENSSL_DEP) src/cpp/util/time.cc: $(OPENSSL_DEP) endif @@ -2739,7 +2737,6 @@ objs/$(CONFIG)/src/cpp/server/server_context_impl.o: objs/$(CONFIG)/src/cpp/server/server_credentials.o: objs/$(CONFIG)/src/cpp/server/server_rpc_handler.o: objs/$(CONFIG)/src/cpp/server/thread_pool.o: -objs/$(CONFIG)/src/cpp/stream/stream_context.o: objs/$(CONFIG)/src/cpp/util/status.o: objs/$(CONFIG)/src/cpp/util/time.o: diff --git a/build.json b/build.json index e3fc12cc66b..3bedddfcd1f 100644 --- a/build.json +++ b/build.json @@ -401,7 +401,6 @@ "src/cpp/proto/proto_utils.h", "src/cpp/server/server_rpc_handler.h", "src/cpp/server/thread_pool.h", - "src/cpp/stream/stream_context.h", "src/cpp/util/time.h" ], "src": [ @@ -421,7 +420,6 @@ "src/cpp/server/server_credentials.cc", "src/cpp/server/server_rpc_handler.cc", "src/cpp/server/thread_pool.cc", - "src/cpp/stream/stream_context.cc", "src/cpp/util/status.cc", "src/cpp/util/time.cc" ], diff --git a/include/grpc++/impl/service_type.h b/include/grpc++/impl/service_type.h index 6e50c43493c..0684f322d8d 100644 --- a/include/grpc++/impl/service_type.h +++ b/include/grpc++/impl/service_type.h @@ -47,6 +47,7 @@ class SynchronousService { class AsynchronousService { public: virtual ~AsynchronousService() {} + virtual RpcService *service() = 0; }; } // namespace grpc diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index ecae429af50..908b38f88e1 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -307,7 +307,7 @@ void PrintHeaderService(google::protobuf::io::Printer *printer, for (int i = 0; i < service->method_count(); ++i) { PrintHeaderServerMethodSync(printer, service->method(i), vars); } - printer->Print("::grpc::RpcService* service();\n"); + printer->Print("::grpc::RpcService* service() override final;\n"); printer->Outdent(); printer->Print( " private:\n" @@ -324,7 +324,7 @@ void PrintHeaderService(google::protobuf::io::Printer *printer, for (int i = 0; i < service->method_count(); ++i) { PrintHeaderServerMethodAsync(printer, service->method(i), vars); } - printer->Print("::grpc::RpcService* service();\n"); + printer->Print("::grpc::RpcService* service() override;\n"); printer->Outdent(); printer->Print( " private:\n" diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index 8d8276ca002..d6bcb9313aa 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -67,6 +67,10 @@ void ServerBuilder::SetThreadPool(ThreadPoolInterface *thread_pool) { std::unique_ptr ServerBuilder::BuildAndStart() { bool thread_pool_owned = false; + if (!async_services_.empty() && !services_.empty()) { + gpr_log(GPR_ERROR, "Mixing async and sync services is unsupported for now"); + return nullptr; + } if (!thread_pool_ && services_.size()) { int cores = gpr_cpu_num_cores(); if (!cores) cores = 4; diff --git a/src/cpp/stream/stream_context.cc b/src/cpp/stream/stream_context.cc deleted file mode 100644 index e4f344dbb93..00000000000 --- a/src/cpp/stream/stream_context.cc +++ /dev/null @@ -1,179 +0,0 @@ -/* - * - * Copyright 2014, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -#include "src/cpp/stream/stream_context.h" - -#include -#include "src/cpp/proto/proto_utils.h" -#include "src/cpp/util/time.h" -#include -#include -#include -#include - -namespace grpc { - -// Client only ctor -StreamContext::StreamContext(const RpcMethod &method, ClientContext *context, - const google::protobuf::Message *request, - google::protobuf::Message *result) - : is_client_(true), - method_(&method), - call_(context->call()), - cq_(context->cq()), - request_(const_cast(request)), - result_(result), - peer_halfclosed_(false), - self_halfclosed_(false) { - GPR_ASSERT(method_->method_type() != RpcMethod::RpcType::NORMAL_RPC); -} - -// Server only ctor -StreamContext::StreamContext(const RpcMethod &method, grpc_call *call, - grpc_completion_queue *cq, - google::protobuf::Message *request, - google::protobuf::Message *result) - : is_client_(false), - method_(&method), - call_(call), - cq_(cq), - request_(request), - result_(result), - peer_halfclosed_(false), - self_halfclosed_(false) { - GPR_ASSERT(method_->method_type() != RpcMethod::RpcType::NORMAL_RPC); -} - -StreamContext::~StreamContext() {} - -void StreamContext::Start(bool buffered) { - if (is_client_) { - // TODO(yangg) handle metadata send path - int flag = buffered ? GRPC_WRITE_BUFFER_HINT : 0; - grpc_call_error error = grpc_call_invoke_old( - call(), cq(), client_metadata_read_tag(), finished_tag(), flag); - GPR_ASSERT(GRPC_CALL_OK == error); - } else { - // TODO(yangg) metadata needs to be added before accept - // TODO(yangg) correctly set flag to accept - GPR_ASSERT(grpc_call_server_accept_old(call(), cq(), finished_tag()) == - GRPC_CALL_OK); - GPR_ASSERT(grpc_call_server_end_initial_metadata_old(call(), 0) == - GRPC_CALL_OK); - } -} - -bool StreamContext::Read(google::protobuf::Message *msg) { - // TODO(yangg) check peer_halfclosed_ here for possible early return. - grpc_call_error err = grpc_call_start_read_old(call(), read_tag()); - GPR_ASSERT(err == GRPC_CALL_OK); - grpc_event *read_ev = - grpc_completion_queue_pluck(cq(), read_tag(), gpr_inf_future); - GPR_ASSERT(read_ev->type == GRPC_READ); - bool ret = true; - if (read_ev->data.read) { - if (!DeserializeProto(read_ev->data.read, msg)) { - ret = false; - grpc_call_cancel_with_status(call(), GRPC_STATUS_DATA_LOSS, - "Failed to parse incoming proto"); - } - } else { - ret = false; - peer_halfclosed_ = true; - } - grpc_event_finish(read_ev); - return ret; -} - -bool StreamContext::Write(const google::protobuf::Message *msg, bool is_last) { - // TODO(yangg) check self_halfclosed_ for possible early return. - bool ret = true; - grpc_event *ev = nullptr; - - if (msg) { - grpc_byte_buffer *out_buf = nullptr; - if (!SerializeProto(*msg, &out_buf)) { - grpc_call_cancel_with_status(call(), GRPC_STATUS_INVALID_ARGUMENT, - "Failed to serialize outgoing proto"); - return false; - } - int flag = is_last ? GRPC_WRITE_BUFFER_HINT : 0; - grpc_call_error err = - grpc_call_start_write_old(call(), out_buf, write_tag(), flag); - grpc_byte_buffer_destroy(out_buf); - GPR_ASSERT(err == GRPC_CALL_OK); - - ev = grpc_completion_queue_pluck(cq(), write_tag(), gpr_inf_future); - GPR_ASSERT(ev->type == GRPC_WRITE_ACCEPTED); - - ret = ev->data.write_accepted == GRPC_OP_OK; - grpc_event_finish(ev); - } - if (ret && is_last) { - grpc_call_error err = grpc_call_writes_done_old(call(), halfclose_tag()); - GPR_ASSERT(err == GRPC_CALL_OK); - ev = grpc_completion_queue_pluck(cq(), halfclose_tag(), gpr_inf_future); - GPR_ASSERT(ev->type == GRPC_FINISH_ACCEPTED); - grpc_event_finish(ev); - - self_halfclosed_ = true; - } else if (!ret) { // Stream broken - self_halfclosed_ = true; - peer_halfclosed_ = true; - } - - return ret; -} - -const Status &StreamContext::Wait() { - // TODO(yangg) properly support metadata - grpc_event *metadata_ev = grpc_completion_queue_pluck( - cq(), client_metadata_read_tag(), gpr_inf_future); - grpc_event_finish(metadata_ev); - // TODO(yangg) protect states by a mutex, including other places. - if (!self_halfclosed_ || !peer_halfclosed_) { - Cancel(); - } - grpc_event *finish_ev = - grpc_completion_queue_pluck(cq(), finished_tag(), gpr_inf_future); - GPR_ASSERT(finish_ev->type == GRPC_FINISHED); - final_status_ = Status( - static_cast(finish_ev->data.finished.status), - finish_ev->data.finished.details ? finish_ev->data.finished.details : ""); - grpc_event_finish(finish_ev); - return final_status_; -} - -void StreamContext::Cancel() { grpc_call_cancel(call()); } - -} // namespace grpc diff --git a/src/cpp/stream/stream_context.h b/src/cpp/stream/stream_context.h deleted file mode 100644 index 8def589841b..00000000000 --- a/src/cpp/stream/stream_context.h +++ /dev/null @@ -1,99 +0,0 @@ -/* - * - * Copyright 2014, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -#ifndef __GRPCPP_INTERNAL_STREAM_STREAM_CONTEXT_H__ -#define __GRPCPP_INTERNAL_STREAM_STREAM_CONTEXT_H__ - -#include -#include -#include - -namespace google { -namespace protobuf { -class Message; -} -} - -namespace grpc { -class ClientContext; -class RpcMethod; - -class StreamContext final : public StreamContextInterface { - public: - StreamContext(const RpcMethod &method, ClientContext *context, - const google::protobuf::Message *request, - google::protobuf::Message *result); - StreamContext(const RpcMethod &method, grpc_call *call, - grpc_completion_queue *cq, google::protobuf::Message *request, - google::protobuf::Message *result); - ~StreamContext(); - // Start the stream, if there is a final write following immediately, set - // buffered so that the messages can be sent in batch. - void Start(bool buffered) override; - bool Read(google::protobuf::Message *msg) override; - bool Write(const google::protobuf::Message *msg, bool is_last) override; - const Status &Wait() override; - void Cancel() override; - - google::protobuf::Message *request() override { return request_; } - google::protobuf::Message *response() override { return result_; } - - private: - // Unique tags for plucking events from the c layer. this pointer is casted - // to char* to create single byte step between tags. It implicitly relies on - // that StreamContext is large enough to contain all the pointers. - void *finished_tag() { return reinterpret_cast(this); } - void *read_tag() { return reinterpret_cast(this) + 1; } - void *write_tag() { return reinterpret_cast(this) + 2; } - void *halfclose_tag() { return reinterpret_cast(this) + 3; } - void *client_metadata_read_tag() { - return reinterpret_cast(this) + 5; - } - grpc_call *call() { return call_; } - grpc_completion_queue *cq() { return cq_; } - - bool is_client_; - const RpcMethod *method_; // not owned - grpc_call *call_; // not owned - grpc_completion_queue *cq_; // not owned - google::protobuf::Message *request_; // first request, not owned - google::protobuf::Message *result_; // last response, not owned - - bool peer_halfclosed_; - bool self_halfclosed_; - Status final_status_; -}; - -} // namespace grpc - -#endif // __GRPCPP_INTERNAL_STREAM_STREAM_CONTEXT_H__