diff --git a/include/grpc++/server.h b/include/grpc++/server.h index 3931d9a1bce..ae86683f0b9 100644 --- a/include/grpc++/server.h +++ b/include/grpc++/server.h @@ -70,15 +70,15 @@ class Server { friend class ServerBuilder; // ServerBuilder use only - Server(ThreadPoolInterface* thread_pool, ServerCredentials* creds); + Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned, ServerCredentials* creds); Server(); // Register a service. This call does not take ownership of the service. // The service must exist for the lifetime of the Server instance. - void RegisterService(RpcService* service); + bool RegisterService(RpcService* service); // Add a listening port. Can be called multiple times. - void AddPort(const grpc::string& addr); + int AddPort(const grpc::string& addr); // Start the server. - void Start(); + bool Start(); void AllowOneRpc(); void HandleQueueClosed(); diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h index f9a40b302d3..8b4c81bc873 100644 --- a/include/grpc++/server_builder.h +++ b/include/grpc++/server_builder.h @@ -57,7 +57,7 @@ class ServerBuilder { // BuildAndStart(). void RegisterService(SynchronousService* service); - void ReigsterAsyncService(AsynchronousService *service); + void RegisterAsyncService(AsynchronousService *service); // Add a listening port. Can be called multiple times. void AddPort(const grpc::string& addr); @@ -78,7 +78,7 @@ class ServerBuilder { std::vector async_services_; std::vector ports_; std::shared_ptr creds_; - ThreadPoolInterface* thread_pool_; + ThreadPoolInterface* thread_pool_ = nullptr; }; } // namespace grpc diff --git a/src/core/support/cpu.h b/include/grpc/support/cpu.h similarity index 100% rename from src/core/support/cpu.h rename to include/grpc/support/cpu.h diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc index 1abdf702e21..ac1b9ddc505 100644 --- a/src/cpp/server/server.cc +++ b/src/cpp/server/server.cc @@ -38,24 +38,20 @@ #include #include #include "src/cpp/server/server_rpc_handler.h" -#include "src/cpp/server/thread_pool.h" #include #include #include #include +#include namespace grpc { -// TODO(rocking): consider a better default value like num of cores. -static const int kNumThreads = 4; - -Server::Server(ThreadPoolInterface *thread_pool, ServerCredentials *creds) +Server::Server(ThreadPoolInterface *thread_pool, bool thread_pool_owned, ServerCredentials *creds) : started_(false), shutdown_(false), num_running_cb_(0), - thread_pool_(thread_pool == nullptr ? new ThreadPool(kNumThreads) - : thread_pool), - thread_pool_owned_(thread_pool == nullptr), + thread_pool_(thread_pool), + thread_pool_owned_(thread_pool_owned), secure_(creds != nullptr) { if (creds) { server_ = @@ -82,31 +78,35 @@ Server::~Server() { } } -void Server::RegisterService(RpcService *service) { +bool Server::RegisterService(RpcService *service) { for (int i = 0; i < service->GetMethodCount(); ++i) { RpcServiceMethod *method = service->GetMethod(i); + if (method_map_.find(method->name()) != method_map_.end()) { + gpr_log(GPR_DEBUG, "Attempt to register %s multiple times", method->name()); + return false; + } method_map_.insert(std::make_pair(method->name(), method)); } } -void Server::AddPort(const grpc::string &addr) { +int Server::AddPort(const grpc::string &addr) { GPR_ASSERT(!started_); - int success; if (secure_) { - success = grpc_server_add_secure_http2_port(server_, addr.c_str()); + return grpc_server_add_secure_http2_port(server_, addr.c_str()); } else { - success = grpc_server_add_http2_port(server_, addr.c_str()); + return grpc_server_add_http2_port(server_, addr.c_str()); } - GPR_ASSERT(success); } -void Server::Start() { +bool Server::Start() { GPR_ASSERT(!started_); started_ = true; grpc_server_start(server_); // Start processing rpcs. ScheduleCallback(); + + return true; } void Server::AllowOneRpc() { @@ -141,8 +141,7 @@ void Server::ScheduleCallback() { std::unique_lock lock(mu_); num_running_cb_++; } - std::function callback = std::bind(&Server::RunRpc, this); - thread_pool_->ScheduleCallback(callback); + thread_pool_->ScheduleCallback(std::bind(&Server::RunRpc, this)); } void Server::RunRpc() { diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index 66e2055af09..8d8276ca002 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -33,13 +33,15 @@ #include +#include #include #include #include +#include "src/cpp/server/thread_pool.h" namespace grpc { -ServerBuilder::ServerBuilder() : thread_pool_(nullptr) {} +ServerBuilder::ServerBuilder() {} void ServerBuilder::RegisterService(SynchronousService *service) { services_.push_back(service->service()); @@ -64,14 +66,27 @@ void ServerBuilder::SetThreadPool(ThreadPoolInterface *thread_pool) { } std::unique_ptr ServerBuilder::BuildAndStart() { - std::unique_ptr server(new Server(thread_pool_, creds_.get())); + bool thread_pool_owned = false; + if (!thread_pool_ && services_.size()) { + int cores = gpr_cpu_num_cores(); + if (!cores) cores = 4; + thread_pool_ = new ThreadPool(cores); + thread_pool_owned = true; + } + std::unique_ptr server(new Server(thread_pool_, thread_pool_owned, creds_.get())); for (auto *service : services_) { - server->RegisterService(service); + if (!server->RegisterService(service)) { + return nullptr; + } } for (auto &port : ports_) { - server->AddPort(port); + if (!server->AddPort(port)) { + return nullptr; + } + } + if (!server->Start()) { + return nullptr; } - server->Start(); return server; }