diff options
author | Yang Gao <yangg@google.com> | 2015-03-30 10:44:05 -0700 |
---|---|---|
committer | Yang Gao <yangg@google.com> | 2015-03-30 10:44:05 -0700 |
commit | fbf8a37c5eafcb1f9b27d9bdaac73040785dd6b6 (patch) | |
tree | 10211221689f95049b3156d6c153297c23de88cb /src/cpp | |
parent | 1205f6f534412f3e2deb88b86f66ae58b07aab8a (diff) | |
parent | 675de61e4ba46f6910eab9051ea77ee73cdf8c28 (diff) |
Merge branch 'master' into cancel
Diffstat (limited to 'src/cpp')
-rw-r--r-- | src/cpp/client/generic_stub.cc | 51 | ||||
-rw-r--r-- | src/cpp/client/insecure_credentials.cc | 2 | ||||
-rw-r--r-- | src/cpp/client/secure_credentials.cc | 2 | ||||
-rw-r--r-- | src/cpp/common/call.cc | 3 | ||||
-rw-r--r-- | src/cpp/server/secure_server_credentials.cc | 9 | ||||
-rw-r--r-- | src/cpp/server/server.cc | 4 | ||||
-rw-r--r-- | src/cpp/server/server_builder.cc | 18 | ||||
-rw-r--r-- | src/cpp/server/thread_pool.cc | 45 | ||||
-rw-r--r-- | src/cpp/server/thread_pool.h | 2 |
9 files changed, 96 insertions, 40 deletions
diff --git a/src/cpp/client/generic_stub.cc b/src/cpp/client/generic_stub.cc new file mode 100644 index 0000000000..3bf7bdf45f --- /dev/null +++ b/src/cpp/client/generic_stub.cc @@ -0,0 +1,51 @@ +/* + * + * Copyright 2015, 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 <grpc++/generic_stub.h> + +#include <grpc++/impl/rpc_method.h> + +namespace grpc { + +// begin a call to a named method +std::unique_ptr<GenericClientAsyncReaderWriter> GenericStub::Call( + ClientContext* context, const grpc::string& method, + CompletionQueue* cq, void* tag) { + return std::unique_ptr<GenericClientAsyncReaderWriter>( + new GenericClientAsyncReaderWriter( + channel_.get(), cq, RpcMethod(method.c_str()), context, tag)); +} + + +} // namespace grpc + diff --git a/src/cpp/client/insecure_credentials.cc b/src/cpp/client/insecure_credentials.cc index f3ca430bd4..8945b038de 100644 --- a/src/cpp/client/insecure_credentials.cc +++ b/src/cpp/client/insecure_credentials.cc @@ -31,8 +31,6 @@ * */ -#include <string> - #include <grpc/grpc.h> #include <grpc/support/log.h> diff --git a/src/cpp/client/secure_credentials.cc b/src/cpp/client/secure_credentials.cc index e3c6637623..d6f9acc675 100644 --- a/src/cpp/client/secure_credentials.cc +++ b/src/cpp/client/secure_credentials.cc @@ -31,8 +31,6 @@ * */ -#include <string> - #include <grpc/grpc_security.h> #include <grpc/support/log.h> diff --git a/src/cpp/common/call.cc b/src/cpp/common/call.cc index 5c26a1ad7c..1599e5117f 100644 --- a/src/cpp/common/call.cc +++ b/src/cpp/common/call.cc @@ -148,7 +148,7 @@ void FillMetadataMap(grpc_metadata_array* arr, // TODO(yangg) handle duplicates? metadata->insert(std::pair<grpc::string, grpc::string>( arr->metadata[i].key, - {arr->metadata[i].value, arr->metadata[i].value_length})); + grpc::string(arr->metadata[i].value, arr->metadata[i].value_length))); } grpc_metadata_array_destroy(arr); grpc_metadata_array_init(arr); @@ -186,6 +186,7 @@ void CallOpBuffer::AddRecvMessage(grpc::protobuf::Message* message) { void CallOpBuffer::AddRecvMessage(ByteBuffer* message) { recv_message_buffer_ = message; + recv_message_buffer_->Clear(); } void CallOpBuffer::AddClientSendClose() { client_send_close_ = true; } diff --git a/src/cpp/server/secure_server_credentials.cc b/src/cpp/server/secure_server_credentials.cc index 88f7a9b1a9..49d69a3fb9 100644 --- a/src/cpp/server/secure_server_credentials.cc +++ b/src/cpp/server/secure_server_credentials.cc @@ -59,9 +59,12 @@ class SecureServerCredentials GRPC_FINAL : public ServerCredentials { std::shared_ptr<ServerCredentials> SslServerCredentials( const SslServerCredentialsOptions& options) { std::vector<grpc_ssl_pem_key_cert_pair> pem_key_cert_pairs; - for (const auto& key_cert_pair : options.pem_key_cert_pairs) { - pem_key_cert_pairs.push_back( - {key_cert_pair.private_key.c_str(), key_cert_pair.cert_chain.c_str()}); + for (auto key_cert_pair = options.pem_key_cert_pairs.begin(); + key_cert_pair != options.pem_key_cert_pairs.end(); + key_cert_pair++) { + grpc_ssl_pem_key_cert_pair p = {key_cert_pair->private_key.c_str(), + key_cert_pair->cert_chain.c_str()}; + pem_key_cert_pairs.push_back(p); } grpc_server_credentials* c_creds = grpc_ssl_server_credentials_create( options.pem_root_certs.empty() ? nullptr : options.pem_root_certs.c_str(), diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc index 9dbf3392d8..bd0a23739c 100644 --- a/src/cpp/server/server.cc +++ b/src/cpp/server/server.cc @@ -248,8 +248,8 @@ bool Server::Start() { // Start processing rpcs. if (!sync_methods_.empty()) { - for (auto& m : sync_methods_) { - m.Request(server_); + for (auto m = sync_methods_.begin(); m != sync_methods_.end(); m++) { + m->Request(server_); } ScheduleCallback(); diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index 58bf9d937f..c5e115f396 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -86,24 +86,26 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() { thread_pool_owned = true; } std::unique_ptr<Server> server(new Server(thread_pool_, thread_pool_owned)); - for (auto* service : services_) { - if (!server->RegisterService(service)) { + for (auto service = services_.begin(); service != services_.end(); + service++) { + if (!server->RegisterService(*service)) { return nullptr; } } - for (auto* service : async_services_) { - if (!server->RegisterAsyncService(service)) { + for (auto service = async_services_.begin(); + service != async_services_.end(); service++) { + if (!server->RegisterAsyncService(*service)) { return nullptr; } } if (generic_service_) { server->RegisterAsyncGenericService(generic_service_); } - for (auto& port : ports_) { - int r = server->AddListeningPort(port.addr, port.creds.get()); + for (auto port = ports_.begin(); port != ports_.end(); port++) { + int r = server->AddListeningPort(port->addr, port->creds.get()); if (!r) return nullptr; - if (port.selected_port != nullptr) { - *port.selected_port = r; + if (port->selected_port != nullptr) { + *port->selected_port = r; } } if (!server->Start()) { diff --git a/src/cpp/server/thread_pool.cc b/src/cpp/server/thread_pool.cc index d3013b806c..80c96111b1 100644 --- a/src/cpp/server/thread_pool.cc +++ b/src/cpp/server/thread_pool.cc @@ -35,28 +35,29 @@ namespace grpc { +void ThreadPool::ThreadFunc() { + for (;;) { + // Wait until work is available or we are shutting down. + std::unique_lock<std::mutex> lock(mu_); + if (!shutdown_ && callbacks_.empty()) { + cv_.wait(lock); + } + // Drain callbacks before considering shutdown to ensure all work + // gets completed. + if (!callbacks_.empty()) { + auto cb = callbacks_.front(); + callbacks_.pop(); + lock.unlock(); + cb(); + } else if (shutdown_) { + return; + } + } +} + ThreadPool::ThreadPool(int num_threads) : shutdown_(false) { for (int i = 0; i < num_threads; i++) { - threads_.push_back(std::thread([this]() { - for (;;) { - // Wait until work is available or we are shutting down. - auto have_work = [this]() { return shutdown_ || !callbacks_.empty(); }; - std::unique_lock<std::mutex> lock(mu_); - if (!have_work()) { - cv_.wait(lock, have_work); - } - // Drain callbacks before considering shutdown to ensure all work - // gets completed. - if (!callbacks_.empty()) { - auto cb = callbacks_.front(); - callbacks_.pop(); - lock.unlock(); - cb(); - } else if (shutdown_) { - return; - } - } - })); + threads_.push_back(std::thread(&ThreadPool::ThreadFunc, this)); } } @@ -66,8 +67,8 @@ ThreadPool::~ThreadPool() { shutdown_ = true; cv_.notify_all(); } - for (auto& t : threads_) { - t.join(); + for (auto t = threads_.begin(); t != threads_.end(); t++) { + t->join(); } } diff --git a/src/cpp/server/thread_pool.h b/src/cpp/server/thread_pool.h index 6225d82a0b..41e2009ff1 100644 --- a/src/cpp/server/thread_pool.h +++ b/src/cpp/server/thread_pool.h @@ -58,6 +58,8 @@ class ThreadPool GRPC_FINAL : public ThreadPoolInterface { bool shutdown_; std::queue<std::function<void()>> callbacks_; std::vector<std::thread> threads_; + + void ThreadFunc(); }; } // namespace grpc |